Applies one decoded Postgres transaction's changes to the ArcadeDB mirror ATOMICALLY with the LSN watermark — the effect-once core of the AshArcadic.Replicant sink.

The atomicity mechanism (Ch1)

apply_transaction/2 wraps the WHOLE apply in Ash.transaction(resources, fn -> ... end) — never a raw Arcadic.transaction/3. The data-layer transaction/4 callback that Ash.transaction/2 routes through is what sets the process marker (:ash_arcadic_tx_marker); AshArcadic.Transaction.resolve_conn opens ONE ArcadeDB session on the first write and every subsequent write/read reuses it, so the mirrored data writes AND the checkpoint upsert commit together in a single session. The resources list MUST be non-empty (it includes the checkpoint resource): a bare Ash.transaction([], fn) returns {:ok, fun.()} WITHOUT engaging the data layer — no marker, no session — so each inner Ash.create!/Ash.bulk_destroy! would auto-commit on its own base connection and a mid-transaction failure would leak partial writes, shattering effect-once. (resources/1 is non-empty by construction; the marker being set inside the fn is proven by execution in the test suite. Crash-atomicity itself is proven live in the sink's integration slice.)

The replay gate (F1 / Ch5)

replay_skip?/2 is is_integer(stored) and lsn <= stored. The is_integer/1 guard is load-bearing: a nil (never-applied) checkpoint MUST apply, not skip — and in Erlang term order a number sorts before every atom, so a guardless lsn <= nil is ALWAYS true and would silently skip the never-applied transaction.

Per-change invariant map

• :insert / :update — MERGE upsert by primary key (Ash.create!(..., upsert?: true)). On an :update whose IDENTITY CHANGED — the primary key OR the resolved tenant (the mirror identity is PK + the tenant discriminator, upsert_identity_keys/2) — the old-(tenant, PK) vertex is destroyed FIRST, then the new row upserted (no ghost row on a PK change, no stale cross-tenant copy on a tenant move).
• :delete — atomic bulk destroy by primary key. A nil primary-key value fails closed (never id == nil, which matches 0 rows and silently "succeeds"); a genuine 0-row match (already-absent) is idempotent :ok.
• :truncate — per on_truncate/1: :halt (default) fails closed; :mirror runs a TENANT-BLIND raw MATCH (n:Label) DETACH DELETE n on the whole label, on the session connection so it is atomic with the surrounding changes and the checkpoint upsert.

A change whose {schema, table} is not a mirror target is ignored (:ok).

Value-free error boundary

apply_transaction/2 ALWAYS returns {:ok, lsn} | {:error, <value-free>} — it never raises and never lets a value-bearing term escape, whether the underlying failure arrived as a returned rollback {:error, reason} (an Ash data-layer write failing INSIDE the transaction rolls back via AshArcadic.Transaction.rollback_throw, which AshArcadic.Transaction.run catches and RETURNS as {:error, reason} — bypassing the per-change rescue), a raised exception, or a caught throw/exit. boundary_error lets only PROVABLY value-free terms cross: an AshArcadic.Replicant.Error (atoms only), or a bare structural AshArcadic.Errors.* data-layer error (value-free by AGENTS.md Rule 4 — extracted from its value-BEARING Ash.Error container, which carries the changeset/query, when wrapped). ANY other or unrecognized term maps to a static :sink_failed (fail closed — an unrecognized error is never trusted to be value-free).

The per-change scrub/3 is the inner layer: it converts a RAISE within one change (a cast/helper/guard) to a value-free AshArcadic.Replicant.Error carrying the correct op, so an op-specific reason (:tenant_required, :truncate_halt, …) survives to the boundary; the underlying exception's contents are never inspected.