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

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

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

@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()}
  | {: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.

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

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

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

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

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