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 writes run inside the host transaction and pass
return_notifications?: true, so Ash bundles any notification into the return
value (never firing a notifier) and the sink discards it — Ash notifiers/pubsub
do NOT fire for mirrored changes. (Ash cannot emit notifications from within a
host-managed transaction anyway; notify?: false is not honored by
single-record create/destroy in Ash 3.x, only return_notifications?.)
Summary
Functions
Last durably-persisted commit LSN for the slot (nil = never), the dedup watermark.
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.
Persist a snapshot batch for ctx.table, upserting by PK. On
first_for_table?, clear the resource's mirror rows in-txn first (redo-safety).
Non-tenant 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. Tenant-scoped (and, defensively,
sensitive) resources apply per-record. Does not advance the checkpoint.
Durably set checkpoint := snapshot_lsn and return it (the snapshot handoff commit).
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).
Functions
@spec checkpoint(map()) :: {:ok, Replicant.lsn() | nil} | {:error, term()}
Last durably-persisted commit LSN for the slot (nil = never), the dedup watermark.
@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.
@spec handle_snapshot(map(), [Replicant.Change.t()], map()) :: :ok | {:error, term()}
Persist a snapshot batch for ctx.table, upserting by PK. On
first_for_table?, clear the resource's mirror rows in-txn first (redo-safety).
Non-tenant 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. Tenant-scoped (and, defensively,
sensitive) resources apply per-record. Does not advance the checkpoint.
@spec handle_snapshot_complete(map(), Replicant.lsn()) :: {:ok, Replicant.lsn()} | {:error, term()}
Durably set checkpoint := snapshot_lsn and return it (the snapshot handoff commit).
@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/2 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.