AshArcadic.Replicant.Apply (AshArcadic v0.2.0)

Copy Markdown View Source

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.

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

config()

@type config() :: %{
  :resolver_index => %{
    required(AshArcadic.Replicant.Resolver.source_key()) => module()
  },
  :checkpoint => module(),
  :slot => String.t(),
  :authorize? => boolean(),
  optional(any()) => any()
}

outcome()

@type outcome() :: {:applied, non_neg_integer()} | {:skipped, integer() | nil}

Functions

apply_change(config, change)

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

apply_transaction(config, transaction)

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