# `Replicant.Assembler`
[🔗](https://github.com/baselabs/replicant/blob/v0.3.1/lib/replicant/assembler.ex#L1)

Assembles a stream of decoded pgoutput messages into a `Replicant.Transaction`
and applies the sink synchronously per transaction (spec §4/§6).

The state machine:
  * `Begin`      — open a transaction buffer.
  * `Relation`   — cache the relation; diff against the cached one and emit/halt
                   on a schema change (spec §9). `:additive` auto-applies;
                   `:destructive` halts fail-closed (or delegates to the sink's
                   optional `handle_schema_change/2`).
  * `Type`       — cache the type name.
  * `Insert/Update/Delete/Truncate` — accumulate a `Change`, casting values via
                   `Replicant.Casting.Types.cast_record/2` and extracting
                   unchanged-TOAST sentinels into `change.unchanged` (spec §7).
                   A row/truncate for a relation that was never cached halts
                   fail-closed (we cannot identify the table — never emit a
                   table-less change that would be checkpointed as success).
  * `Commit`     — attach the transaction-granularity commit LSN; pre-skip if
                   `commit_lsn <= checkpoint()` (spec §2); else call
                   `sink.handle_transaction/1` synchronously and return the LSN.

The Connection (Plan 2) drives this; the Assembler never blocks the keepalive
path itself — that is Plan 2's process-structure concern.

## Telemetry (spec §10, Assembler-owned events)
Emits the value-free events whose lifetime lives in the Assembler:
`[:replicant, :transaction, :assembled]`, `[:replicant, :sink, :committed|:failed]`,
`[:replicant, :schema_change, :additive|:halted]` — metadata scrubbed through the
`Replicant.Telemetry` allowlist. The `:connection`/`:checkpoint` events are the
Connection's (Plan 2).

## Value-free boundary
`handle_message/2` wraps its body in `rescue` AND `catch`: a raise from the
casting path (a malformed numeric/bytea in `Types.cast_record/2`) is scrubbed
into `{:halt, %Error{reason: :decode_failure}}`; a sink raise/throw/exit is
scrubbed by `apply_sink/2` into `{:halt, %Error{reason: :sink_failed}}` — never
inspecting a throw/exit reason, which (e.g. a `GenServer.call` timeout carrying
the transaction) can embed row values (Critical Rule 1). A row/commit message
arriving before any `Begin` halts as `:config_invalid` rather than crashing.

# `buffer`

```elixir
@type buffer() :: %{
  begin_lsn: lsn() | nil,
  xid: non_neg_integer() | nil,
  changes: [Replicant.Change.t()],
  messages: [Replicant.Decoder.Messages.Message.t()],
  byte_size: non_neg_integer()
}
```

# `lsn`

```elixir
@type lsn() :: Replicant.lsn()
```

# `stream_buf`

```elixir
@type stream_buf() :: %{
  changes: [{non_neg_integer(), Replicant.Change.t()}],
  messages: [Replicant.Decoder.Messages.Message.t()],
  byte_size: non_neg_integer(),
  resident_bytes: non_neg_integer(),
  spilled_bytes: non_neg_integer(),
  spilled_by_subxid: %{required(non_neg_integer()) =&gt; non_neg_integer()},
  aborted: MapSet.t(non_neg_integer()),
  spill: Replicant.Spill.handle() | nil,
  seq: non_neg_integer()
}
```

# `t`

```elixir
@type t() :: %Replicant.Assembler{
  batch: keyword() | nil,
  batch_count: non_neg_integer(),
  batch_spill_paths: [String.t()],
  batch_txns: [Replicant.Transaction.t()],
  checkpoint_writer: (lsn() -&gt; :ok | {:error, term()}) | nil,
  current_stream_xid: non_neg_integer() | nil,
  last_buffered_changes: [Replicant.Change.t()] | :spilled,
  lib_checkpoint: lsn() | nil,
  max_concurrent_txns: pos_integer() | nil,
  max_inflight_lag: pos_integer() | nil,
  mode: :sink_owned | :lib,
  ordinal: non_neg_integer(),
  pending_lsn: lsn() | nil,
  projected: %{required(non_neg_integer()) =&gt; [Replicant.Change.Column.t()]},
  relations: %{
    required(non_neg_integer()) =&gt; Replicant.Decoder.Messages.Relation.t()
  },
  sink: module(),
  slot_name: String.t() | nil,
  spill: keyword() | nil,
  spill_fault: Replicant.Error.t() | nil,
  spilled_total: non_neg_integer(),
  stream_floor: lsn() | nil,
  stream_txns: %{required(non_neg_integer()) =&gt; stream_buf()},
  txn: buffer() | nil
}
```

# `batch_pending?`

```elixir
@spec batch_pending?(t()) :: boolean()
```

True when a batch is open (buffered, not yet flushed) — the flush-timer relevance guard (spec §9).

# `flush_batch`

```elixir
@spec flush_batch(t(), atom()) ::
  {:ok, lsn(), t()} | {:error, Replicant.Error.t(), t()} | :empty
```

Flush the open batch (spec §7): write ONE store checkpoint at the batch's highest committed
LSN via the injected `checkpoint_writer` (which carries the bounded retry), advance the
in-memory watermark to it (keeping `lib_checkpoint` == the durable store checkpoint, spec §9),
and reset the accumulators. Emits `[:replicant, :checkpoint_store, :batch_flushed]` on success
(value-free: slot_name + counts + the trigger `reason`). A write fault returns
`{:error, %Error{}, t()}` so the AssemblerServer halts fail-closed WITHOUT advancing — the whole
batch re-delivers on restart (dup ≤ batch, never loss). `:empty` when no batch is open (a stale
flush-timer fire — the pending_lsn is nil).

# `handle_message`

```elixir
@spec handle_message(
  t(),
  struct()
) ::
  {:ok, t()}
  | {:transaction, Replicant.Transaction.t(), lsn(), t()}
  | {:skipped, lsn(), t()}
  | {:schema_change, Replicant.SchemaChange.t(), t()}
  | {:buffered, t()}
  | {:flush, atom(), t()}
  | {:flush_before_message, atom(), t(), Replicant.Decoder.Messages.Message.t()}
  | {:message_delivered, lsn(), t()}
  | {:halt, term(), t()}
```

Handle one decoded message. Returns:
  * `{:ok, t()}` — accumulated, no boundary crossed.
  * `{:transaction, Transaction.t(), lsn(), t()}` — Commit, sink committed.
  * `{:skipped, lsn(), t()}` — Commit but `commit_lsn <= checkpoint` (watermark skip).
  * `{:buffered, t()}` — lib+batch: applied to the sink, checkpoint pending (no ack yet).
  * `{:flush, reason, t()}` — lib+batch: the count/span cap tripped; the AssemblerServer must flush.
  * `{:schema_change, SchemaChange.t(), t()}` — additive schema change applied.
  * `{:halt, SchemaChange.t() | term(), t()}` — destructive schema change, sink
    failure, an unidentifiable-relation row, or a value-bearing raise (e.g.
    casting a malformed numeric) — fail-closed, value-free.

`handle_message/2` is itself the value-free boundary for the casting path: the
vendored `Types.cast_record/2` raises `ArgumentError` on malformed numerics /
bytea (a corrupted stream), and those exceptions embed row bytes. The public
wrapper scrubs any such raise into `{:halt, %Error{reason: :decode_failure}}`
and any stray throw/exit (defense-in-depth) likewise, never inspecting the
reason (Critical Rule 1) — symmetric to `Replicant.Decoder.decode/1`.

# `last_buffered_changes`

```elixir
@spec last_buffered_changes(t()) :: [Replicant.Change.t()] | :spilled
```

The LAST lib+batch `buffer_txn`'s changes (a `%Change{}` list) so the AssemblerServer can DROP-
FILTER colliding snapshot-chunk rows, or `:spilled` when that txn's `changes` was a lazy spill-
backed Enumerable and unenumerable (server taints every open table → re-read instead, spec §2/§4).

# `new`

```elixir
@spec new(
  module(),
  keyword()
) :: t()
```

Create an assembler bound to `sink` (a module implementing `Replicant.Sink`).
`opts` selects the checkpoint mode:

  * (default) sink-owned — the sink returns its own checkpoint; `checkpoint/0`
    is the watermark read live per Commit;
  * `mode: :lib` — the library owns the checkpoint: `:checkpoint_writer` (a
    `(lsn -> :ok | {:error, _})`) persists it after the sink, `:lib_checkpoint`
    seeds the in-memory watermark used by the pre-skip, and `:slot_name` labels
    the value-free `[:replicant, :checkpoint_store, :failed]` telemetry (spec §10).

# `observe_bytes`

```elixir
@spec observe_bytes(t(), non_neg_integer()) :: t()
```

Accumulate the raw WAL payload byte-size of the message about to be handled into
the open transaction buffer, for the `byte_size` metadata on
`[:replicant, :transaction, :assembled]` (spec §10). `Replicant.AssemblerServer`
calls this with `byte_size(payload)` before each `handle_message/2`. It is a
no-op before any `Begin` (bytes arriving outside a transaction — a lone
Relation/Type/Origin — have no buffer to attribute to), so a stray pre-Begin
payload never crashes here.

# `put_stream_floor`

```elixir
@spec put_stream_floor(t(), lsn()) :: t()
```

Record the per-stream floor (the Connection's first-frame `wal_end` — where PG actually began
streaming). It is the cold-start component of the span-cap base `max(lib_checkpoint, stream_floor)`
(spec §7): on a fresh slot lib_checkpoint is 0, so measuring the batch's lag from the stream floor
(not absolute 0) is what keeps a large-absolute first-txn LSN from spuriously span-flushing.

# `reset_batch`

```elixir
@spec reset_batch(t()) :: t()
```

Discard any open batch — clear the accumulators without checkpointing (spec §9). Used on a
mid-stream Connection reconnect re-seed: the un-checkpointed batch re-streams from the durable
checkpoint and re-buffers as a FRESH batch, so a transient reconnect matches the crash/stop→resume
dup model (bounds dup to one batch per reconnect; stale accumulators would misalign flush
boundaries). Never writes a checkpoint — loss=0 holds by re-delivery.

# `reset_streams`

```elixir
@spec reset_streams(t()) :: t()
```

Discard all in-progress streamed transactions (on reconnect); loss=0 by re-stream (spec §9). Any
open spill file is deleted first (a reconnect re-streams from the durable checkpoint, so a stale
spill file must not survive — spec §5 reset cleanup), and `spilled_total` resets to 0.

---

*Consult [api-reference.md](api-reference.md) for complete listing*