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:updatewhose 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. Anilprimary-key value fails closed (neverid == nil, which matches 0 rows and silently "succeeds"); a genuine 0-row match (already-absent) is idempotent:ok.:truncate— peron_truncate/1::halt(default) fails closed;:mirrorruns a TENANT-BLIND rawMATCH (n:Label) DETACH DELETE non 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.
Summary
Functions
Apply a single change under config. A change whose {schema, table} is not a mirror
target is ignored (:ok). Called per change, in delivery order, inside
apply_transaction/2's session. Raises a VALUE-FREE AshArcadic.Replicant.Error on any
failure so the surrounding transaction rolls back (the fail-closed effect-once contract).
Apply one decoded transaction's changes atomically with the watermark advance.
Types
@type outcome() :: {:applied, non_neg_integer()} | {:skipped, integer() | nil}
Functions
@spec apply_change(config(), Replicant.Change.t()) :: :ok
Apply a single change under config. A change whose {schema, table} is not a mirror
target is ignored (:ok). Called per change, in delivery order, inside
apply_transaction/2's session. Raises a VALUE-FREE AshArcadic.Replicant.Error on any
failure so the surrounding transaction rolls back (the fail-closed effect-once contract).
@spec apply_transaction(config(), Replicant.Transaction.t()) :: {:ok, integer() | nil} | {:error, term()}
Apply one decoded transaction's changes atomically with the watermark advance.
Routes through Ash.transaction/2 (Ch1) so the mirrored writes and the checkpoint
upsert commit in ONE ArcadeDB session. Skips a replayed transaction (F1 gate) and
otherwise applies every change in delivery order, then advances the checkpoint to
commit_lsn. Returns {:ok, applied_lsn} (the Sink's {:ok, lsn} contract) or
{:error, reason} from a rolled-back transaction.