# `AshReplicant.Sink.Impl`
[🔗](https://github.com/baselabs/ash_replicant/blob/v1.2.0/lib/ash_replicant/sink/impl.ex#L1)

The config-parameterized sink implementation the `use AshReplicant.Sink`
macro delegates to. `handle_transaction/2` is the effect-once core: one
`Repo.transaction` wrapping {dedup-check → single-pass apply → checkpoint
upsert}, value-free and fail-closed.

Mirrored creates and SCD2 updates run inside the host transaction with
`return_notifications?: true`, so Ash bundles notifications into the return
value and the sink discards them. Mirrored bulk destroys use Ash's default
`notify?: false`. Ash notifiers/pubsub therefore do not fire for mirrored
changes; the mechanisms differ by action path.

Suppression is DISPATCH-only (U3/D2): Ash runs a notifier's `load/2`
dependency pre-load read inside the delivery path regardless of any notify
gate — including the snapshot `bulk_create` under `return_records?: false`
(the sink passes `return_notifications?: true`, keeping the pre-load
alive). A load-carrying notifier therefore executes host reads inside the
admitted transaction; the destination manifest requires such notifiers to
declare `DestinationParticipant` (the `:notifier` kind), admitting those
reads into the graph — the same trust model as `tenant_mfa`.

# `checkpoint`

```elixir
@spec checkpoint(map()) :: {:ok, Replicant.lsn() | nil} | {:error, term()}
```

Last durably-persisted commit LSN for the slot (`nil` = never), the dedup watermark.

# `handle_batch`

```elixir
@spec handle_batch(map(), [Replicant.Transaction.t()]) ::
  {:ok, Replicant.lsn() | nil} | {:error, term()}
```

Deliver a BATCH of committed transactions as ONE atomic unit (C2/ADR-0016,
the eighth sink boundary body). The transactions arrive in ascending
`commit_lsn` order; the body applies them and their transactional messages
through the SAME single-pass core as `handle_transaction/2`, then persists
`checkpoint := the batch's highest commit_lsn` AFTER all effects — one
destination transaction, one watermark write, effect-once (dup = 0) across
a mid-batch teardown. Any frontier at/below the locked watermark skips (the
whole batch when the highest LSN is stale). A fault in ANY transaction rolls
back EVERY transaction's effects and advances nothing — fail-closed,
value-free (rescue AND catch, like every boundary body). A lazy spilled
transaction's `changes` is enumerated exactly once, never materialized.

# `handle_message`

```elixir
@spec handle_message(map(), Replicant.Decoder.Messages.Message.t(), map()) ::
  :ok | {:error, term()}
```

Deliver a NON-TRANSACTIONAL logical-decoding message standalone (the
seventh sink boundary body, C1/ADR-0015). Routing is fail-closed: an
explicitly ignored prefix acknowledges (watermark advanced, no effect); an
unknown prefix halts (`:message_prefix_unmapped`); a database-local route
applies effect+claim+watermark in ONE destination transaction; an external
peer route applies through AshOnetime's three-state recovery and advances
the watermark only after a finalized/replayed success. Value-free on every
fault shape (rescue AND catch) with the sink's own `:halted` event.

# `handle_schema_change`

```elixir
@spec handle_schema_change(map(), Replicant.SchemaChange.t(), map()) ::
  :ok | {:error, term()}
```

Accept or decline a schema change. An `:additive` change auto-applies; a
`:destructive` change on a resource whose `on_schema_change` is
`:halt_destructive` (default) halts fail-closed. The context map is not
value-inspected. Unmapped tables use the behaviour default.

# `handle_slot_origin`

```elixir
@spec handle_slot_origin(map(), Replicant.lsn(), map()) :: :ok | {:error, term()}
```

Admit the replication slot's consistent-point origin for a GO-FORWARD append
sink (ADR-0018 §5). Generated only on such sinks, and invoked on every
connect and reconnect BEFORE `START_REPLICATION` — after
`handle_session_identity/3` has bound the checkpoint row, so the row is
always present here.

The first admitted activation persists the callback origin as the log's
IMMUTABLE floor: a new slot supplies its `CREATE_REPLICATION_SLOT`
consistent point, a reused slot its effective `START_REPLICATION` origin.
No completeness claim covers data below that floor.

Every LATER origin is a moving resume fact, never a replacement floor, and it
has to tie out against what the log durably holds:

  * a slot CREATED this session (`reused?: false`) arriving at a log that
    already claims a floor means the previous slot is gone, so PostgreSQL no
    longer retains the WAL between the durable frontier and this new
    consistent point — it halts `:append_origin_gap` before readiness;
  * an APPENDED event above the durable checkpoint means a torn write, since
    the append and the checkpoint commit in one transaction — it halts
    `:append_frontier_divergent` rather than resuming over it.

Both halts are value-free; the origin and the identity never render.

# `handle_snapshot`

```elixir
@spec handle_snapshot(map(), [Replicant.Change.t()], map()) :: :ok | {:error, term()}
```

Persist a snapshot batch for `ctx.table`. Does not advance the checkpoint.

**No callback clears a resource** (S02 / ADR-0017). The pre-S02 code wiped
the whole mirror on `first_for_table?` for redo-safety; that repeated every
committed host business effect on any retry and could erase a stream-applied
row. Stale rows are now retired at fenced completion instead, per tenant
scope, through the host's own retire action — and only for a resource that
opted into `snapshot_provenance`. `first_for_table?` authorizes no deletion
and carries no attempt identity.

A `snapshot_provenance true` resource routes to
`AshReplicant.Snapshot.Rows`: the batch's rows are compared against their
stored fingerprints under the run's bound attempt, and an unchanged row is
marked rather than re-run. Everything else keeps the pre-S02 apply. Plain
SCD1 non-tenant, non-sensitive resources use a bulk upsert; the load-bearing
fail-closed guard is the `case result.status` check — anything other than
`:success` (including the default-options `:partial_success`) rolls the
snapshot transaction back, so a failing row is never silently dropped.
`stop_on_error?: true` is a defensible early-stop on top of that, not the loss
guard. Sensitive, tenant-scoped, OR SCD2 resources apply per-record — SCD2
stamps the batch's snapshot LSN onto each change so each version opens at
`valid_from_lsn = snapshot_lsn`.

Incremental chunks persist their exact opaque progress with the row effects
and durable ordinal cursor. The dedicated empty completion callback retires
unseen rows and stores its token-hash replay fence without regressing the
stream watermark.

# `handle_snapshot_complete`

```elixir
@spec handle_snapshot_complete(map(), Replicant.lsn()) ::
  {:ok, Replicant.lsn()} | {:error, term()}
```

Durably set `checkpoint := snapshot_lsn` and return it (the snapshot handoff commit).

# `handle_transaction`

```elixir
@spec handle_transaction(map(), Replicant.Transaction.t()) ::
  {:ok, Replicant.lsn()} | {:error, term()}
```

Persist the transaction's changes AND the checkpoint atomically; skip if
`commit_lsn <= checkpoint`. Returns `{:ok, commit_lsn}` or a value-free
`{:error, %AshReplicant.Error{}}` (the pipeline halts fail-closed and
re-delivers on resume).

`Apply.apply_change/3` RAISES on failure, so a failing change propagates out of
`Repo.transaction` (Ecto rolls back, then re-raises) and lands on the outer
`rescue` — NOT the `{:error, _}` branch of the result match. Both halt paths
route through `halt/2`, so `:halted` telemetry fires on the real raise path too.

# `snapshot_progress`

```elixir
@spec snapshot_progress(map()) ::
  {:ok, binary() | nil | :backfill_pending} | {:error, term()}
```

Read or prepare the durable incremental-snapshot progress authority.

---

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