AshReplicant

An Ash Framework adapter for replicant — the framework-agnostic Postgres CDC consumer. Mirrors a source Postgres database's committed streaming transactions into AshPostgres resources with durable effect-once semantics, resolving resource, tenant, and classification in the Ash layer while keeping replicant tenant-blind.

AshReplicant is the "ash_postgres of replicant": define Ash resources backed by a Postgres source's CDC stream, with multitenancy and sensitive-data encryption verification enforced through host Ash actions. Validations, changes, AshCloak hooks, and multitenancy run; the sink uses authorize?: false, so host policies are not re-gated. It executes through the replicant client (the transport — the "postgrex of CDC").

Status: v1.2.0 — stable public API (ADR-0023). Effect-once mirroring with fail-closed multitenancy (compile-time verified), SCD2 history mirroring, and AshCloak integration. Working rules are in AGENTS.md — read it first. A fuller project charter (architecture, scope, and the resolved effect-once model) is tracked at docs/CHARTER.md. Only the /docs/superpowers/ lifecycle artifacts (specs, plans, handoffs) are local-only.

Layering

Ash core        multitenancy DSL, policies, the tenant concept

   │

AshReplicant ← HERE   Ash resource extension: tenant routing, sensitive verification,

   │                   resource mapping, mirror actions

   │

replicant       Postgres logical replication (pgoutput), WAL ordering and delivery

   │

Postgres        logical decoding output (pgoutput protocol)

Multitenancy lives here, not in replicant — exactly as ash_postgres (not postgrex) owns schema-based tenancy. This split is verified by the separate Replicant.Sink behaviour and the dual library structure.

Installation

Add ash_replicant to your dependencies in mix.exs:

# mix.exs

{:ash_replicant, "~> 1.2.0"}

It pulls in replicant (the CDC transport) as a transitive dependency.

Supported foundation

The current 1.2.0 release baseline is built and tested with:

The Ash lower bound excludes known-vulnerable patches, and the upper bound excludes Ash 4 prereleases. The AshOnetime ceiling tracks its 1.0 surface freeze — 2.0 is its next reserved break. A host whose own AshOnetime store was installed before 1.1.0 must run AshOnetime's mix ash_onetime.gen.logical_partitions upgrade before serving message routes on 1.1+. AshOnetime protects admitted local auxiliary actions that need a WAL replay guard and is also the dedup mechanism for logical messages (C1 / ADR-0015): a standalone pg_logical_emit_message routes by prefix to a protected create action whose claim is keyed on source+slot+LSN with a versioned host-keyed content digest as the fingerprint. It does not replace the durable commit-LSN checkpoint used for transaction replay and resume.

See ADR-0002 for the dependency decision and ADR-0003 for the release-evidence contract. ADR-0005 records the Replicant 1.x compatibility and release-order contract.

Install with Igniter

mix igniter.install ash_replicant

That adds the dependency and runs mix ash_replicant.install, which generates the domain, checkpoint, sink, and pipeline supervisor; registers the domain; supervises the pipeline; imports AshReplicant's formatter metadata; and queues mix ash.codegen install_ash_replicant for the checkpoint migration. In an app that already has ash_replicant as a dependency, run the installer directly:

mix ash_replicant.install --repo MyApp.Repo --slot shop_orders

--repo selects among several discovered AshPostgres repos; when none exists, generate one with mix ash_postgres.install first. --slot defaults to <otp_app>_replicant. --domain, --checkpoint, --sink, and --pipeline rename individual artifacts.

It writes no connection, publication, source identity, or key material. Those are operator facts, and a plausible-looking placeholder is worse than an absent one — so the generated pipeline supervises nothing until you configure it, and a fresh install compiles and boots as a no-op. Re-running the installer over an installed project changes nothing.

It stops rather than guess. Malformed module names; an illegal slot name; a missing, ambiguous, unknown, or non-AshPostgres repo; incomplete project facts; a module it did not generate at a target name; an unreadable existing binding; or a checkpoint, sink, or pipeline bound to a different identity each stop the install — writing nothing — with a structural message naming the resolving flag. Re-keying a live sink onto a different slot would abandon its durable checkpoint row and re-deliver from the new slot's position; that is exactly the kind of quiet, expensive wrongness the installer refuses to perform silently.

Igniter is an optional dependency. Without it, mix ash_replicant.install prints the instruction to add it, and the manual path below reaches the identical contract.

Manual installation

Everything the installer writes, by hand. These four modules are the whole generated surface — no hidden state, no package-owned code:

# lib/my_app/replicant.ex

defmodule MyApp.Replicant do

  use Ash.Domain,

    otp_app: :my_app



  resources do

    resource MyApp.Replicant.Checkpoint

  end

end



# lib/my_app/replicant/checkpoint.ex

defmodule MyApp.Replicant.Checkpoint do

  use AshReplicant.Checkpoint,

    repo: MyApp.Repo,

    domain: MyApp.Replicant

end



# lib/my_app/replicant/sink.ex

defmodule MyApp.Replicant.Sink do

  use AshReplicant.Sink,

    repo: MyApp.Repo,

    domains: [],

    checkpoint_resource: MyApp.Replicant.Checkpoint,

    slot_name: "my_app_replicant"

end



# lib/my_app/replicant/pipeline.ex

defmodule MyApp.Replicant.Pipeline do

  use AshReplicant.Pipeline,

    otp_app: :my_app,

    sink: MyApp.Replicant.Sink

end

Then add :ash_replicant to the formatter's existing import_deps, preserving every existing formatter entry; register and supervise the generated modules; and generate the checkpoint migration:

# .formatter.exs — merge into the existing list; keep every other key

[

  import_deps: [:ash, :ash_postgres, :ash_replicant],

  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]

]

# config/config.exs — register the domain

config :my_app, ash_domains: [MyApp.Replicant]

# lib/my_app/application.ex — supervise the pipeline

children = [

  MyApp.Repo,

  MyApp.Replicant.Pipeline

]

mix ash.codegen install_ash_replicant

mix ecto.migrate

The checkpoint migration comes from your own resource snapshots, so it stays in step with the generated resource instead of drifting from a shipped template.

That is the complete equivalent; a test ties this block to the installer's actual output, so the two cannot drift.

After either path

Four steps remain, and each needs a fact only the operator has:

  1. Configure the pipeline. Until this exists, MyApp.Replicant.Pipeline supervises nothing (see AshReplicant.Pipeline):

    config :my_app, MyApp.Replicant.Pipeline,

      connection: [hostname: "standby.example.com", database: "source_db"],

      publication: "shop_orders_pub",

      source_identity: [system_identifier: "7378697629483820647", database: "source_db"],

      go_forward_only: true

    A configuration that is present but missing :connection, :publication, or :source_identityraises rather than supervising nothing — supervising nothing would be a silent outage.

  2. Mark the resources you mirror with the AshReplicant.Resource extension and list their domains in the sink's domains (see Quick Start step 3). Every published table must be mapped or explicitly ignored, or the pipeline refuses to start.

  3. Set ALTER TABLE <table> REPLICA IDENTITY FULL on the SOURCE database for every tenant-scoped, SCD2-with-non-PK-business-key, or append-log source table.

  4. Add the optional key material only if you use the features that need it: :message_digest_keys for logical-message routing (C1) and :snapshot_provenance_keys for snapshot provenance (S02).

Quick Start

The concepts behind what the installer generates, plus the parts only you can write. Steps 1, 2, and 4's supervision are what mix ash_replicant.install produces; step 3 is the modelling decision it deliberately leaves to you.

1. Define the checkpoint resource

defmodule MyApp.ReplicantCheckpoint do

  use AshReplicant.Checkpoint,

    repo: MyApp.Repo,

    domain: MyApp.Domain

end

This generates an AshPostgres resource backed by the ash_replicant_checkpoints table: one row per replication SOURCE and slot, keyed by (source_system_id, source_database, slot_name) from the actual replication session's identity, carrying the durable commit LSN watermark, the recorded session timeline, and the canonical contract manifest with its fingerprint (ADR-0007). The sink binds the row on every connect before any checkpoint read, admits under a FOR UPDATE row lock, and advances the watermark monotonically.

The checkpoint is an internal watermark — nothing outside the sink should read or write it. The generated resource is default-deny: it carries Ash.Policy.Authorizer with an empty policy set, which forbids every external actor on every action — even on a wire surface (JSON:API, MCP) you add later. To grant specific access, declare your own policies:

defmodule MyApp.ReplicantCheckpoint do

  use AshReplicant.Checkpoint,

    repo: MyApp.Repo,

    domain: MyApp.Domain



  policies do

    default_access_type :strict



    policy always() do

      authorize_if MyApp.Checks.SystemActor

    end

  end

end

The sink reads and upserts the checkpoint with authorize?: false, so it bypasses policy — effect-once is unaffected whatever you declare, including nothing (the default). Hosts that already front the resource with their own authorization can reproduce the earlier unguarded shape with authorizers: [] (ADR-0014).

Upgrading 0.4.0's slot-only shape to 1.0.0: use the guarded package task, mix ash_replicant.upgrade 0.4.0 1.0.0. It requires one explicit source-identity binding per configured sink, classifies the live destination read-only before it writes host source, and generates the atomic checkpoint bridge plus its exact resource snapshot. It never infers ownership from a slot-only row. Follow the upgrade and rollback procedure; do not use the former capture/delete/adopt sequence.

2. Define the sink

defmodule MyApp.ReplicantSink do

  use AshReplicant.Sink,

    repo: MyApp.Repo,

    domains: [MyApp.Shop, MyApp.Billing],

    checkpoint_resource: MyApp.ReplicantCheckpoint,

    slot_name: "shop_orders"

end

Key:slot_name is baked into the sink — it is the single source of truth for the replication slot name and is used to key the resolved index. It is NOT a start_link option.

The sink also carries the optional logical-message routing surface (C1 / ADR-0015):

use AshReplicant.Sink,

  repo: MyApp.Repo,

  domains: [MyApp.Shop, MyApp.Messaging],

  checkpoint_resource: MyApp.ReplicantCheckpoint,

  slot_name: "shop_orders",

  message_routes: [{"mail", MyApp.MailOutbox, :record}],

  ignored_message_prefixes: ["telemetry_noise"]

A routed action is a create action protected by the closed AshOnetime message profile (its claim dedups standalone re-delivery; a transactional message rides its transaction and interleaves with the row changes by ordinal). The pipeline starts with messages: true automatically; an unknown prefix halts fail-closed. Set Application.put_env(:ash_replicant, :message_digest_keys, [{1, "<32-byte secret>"}]) (host-keyed content digests; retain old versions when rotating).

3. Mark source resources with the extension

defmodule MyApp.Shop.Order do

  use Ash.Resource,

    domain: MyApp.Shop,

    data_layer: AshPostgres.DataLayer,

    extensions: [AshReplicant.Resource]  # ← HERE



  postgres do

    table "orders"

    repo MyApp.Repo

  end



  replicant do

    source_table "orders"

    source_schema "public"

    tenant_attribute :org_id

    sensitive [:pan, :cvv]

    skip [:internal_field]

    on_truncate :mirror

    on_schema_change :halt_destructive

    upsert_identity :unique_pk

  end



  attributes do

    attribute :id, :uuid, primary_key?: true, public?: true

    attribute :org_id, :uuid, public?: true  # Tenant column; resolved per row and passed as tenant:

    attribute :amount, :decimal, public?: true

    attribute :pan, :binary, public?: true    # Sensitive: binary storage; stored as-is (host-managed encryption unless this resource also uses AshCloak)

    attribute :cvv, :binary, public?: true    # Sensitive: binary storage; stored as-is (host-managed)

    attribute :internal_field, :string, public?: true  # Skipped; not mirrored from source

  end



  actions do

    # The extension generates NO action. The mirror writes through THIS resource's

    # own primary `:create` action (as an upsert) and its `:destroy` action, so you

    # must define them. `create: :*` gives a primary create accepting all public

    # attributes — the upsert target; `:destroy` handles mirrored deletes/truncate.

    defaults [:read, :destroy, create: :*, update: :*]

  end



  identities do

    identity :unique_pk, [:id]

  end

end

About the DSL:

History (SCD2). By default a resource mirrors current state (history_strategy :scd1 — upsert / destroy). Opt a resource into validity-windowed SCD2 history (close-current + insert-version into a host-defined version table, instead of overwriting) with history_strategy :scd2:

replicant do

  source_table "orders"

  history_strategy :scd2

  history_business_key [:order_id]   # source natural key (composite supported)

  upsert_identity :version_key       # identity keys: [:order_id, :valid_from_lsn]

end

See usage-rules.md (“SCD2 history mode”) for the full version-table contract: the surrogate PK, the valid_from_lsn / valid_to_lsn window columns, the :close_version action, the partial-unique-open index, on_truncate :close, and the REPLICA IDENTITY FULL precondition for a non-PK business key.

Snapshot provenance. A snapshot retry must not repeat a host business effect for a row that did not change — converging to the same final state is not enough when a create, destroy, or SCD2 close carries an append-only effect. Opt a resource in with snapshot_provenance true and it stores a keyed fingerprint of the values its host action was given, plus a marker for the attempt that last saw it:

replicant do

  source_table "orders"

  snapshot_provenance true

end



attributes do

  attribute :replica_fingerprint, :binary, public?: false, writable?: false

  attribute :replica_seen_attempt, :binary, public?: false, writable?: false

end



actions do

  update :replicant_mark_seen do

    public? false

    accept []

    require_atomic? false

    change AshReplicant.Snapshot.MarkSeen

  end



  destroy :replicant_retire_unseen do

    public? false

  end

end

A compile-time verifier rejects any action that accepts either attribute or declares an argument named for one, and rejects MarkSeen globally or on any action other than the configured private mark action, so provenance cannot be forged; fingerprint keys come from :ash_replicant, :snapshot_provenance_keys and are preflighted at activation.

On a whole-table (V1) retry the adapter binds one random attempt to the pipeline owner's delivery run. In incremental mode, snapshot_progress/0 arms the durable attempt before Replicant can start the reader and stream, returning :backfill_pending until the first chunk token commits; every chunk commits its exact opaque progress, authenticated progress hash, destination effects, membership markers, and ordinal cursor together. Streaming inserts/updates during the backfill carry the same marker. Both modes mark unchanged rows instead of re-running the host action and retire only managed open rows the attempt never saw. Incremental completion stores matching progress/completion token hashes as a permanent replay fence before returning, so redelivery after a later stream write or an admitted-contract deployment cannot scan, retire, or brick streaming.

No snapshot callback clears a resource. The pre-1.0 whole-resource DELETE on first_for_table? is gone: it repeated every committed host business effect on a retry. A snapshot-backed resource that does not opt into snapshot_provenance now keeps rows the source has dropped — opting in is what restores retirement.

Under strategy :context multitenancy the resource must also declare snapshot_tenant_scope_action, a private generic action returning the retained tenant contexts — there is no discriminator column to enumerate, and a partial enumeration would silently under-retire. See usage-rules.md (“Snapshot provenance and retirement”) for the full contract, how a V1 retry behaves, key rotation, and the migration path.

4. Start the pipeline

AshReplicant.start_link(

  sink: MyApp.ReplicantSink,

  connection: [hostname: "standby.example.com", database: "source_db"],

  publication: "shop_orders_pub",

  source_identity: [system_identifier: "7378697629483820647", database: "source_db"],

  go_forward_only: true,

  snapshot: false,

  census: [interval_ms: 60_000, jitter_ratio: 0.1, timeout_ms: 10_000],

  max_inflight_lag: 64 * 1024 * 1024,

  max_command_retries: 5,

  failover: false

)

Key points:

Operator diagnosis — preflight and doctor

Two commands answer may this start? and what is the state of this deployment? without touching anything:

# Before the first activation — no checkpoint is read, so this is correct on a

# fresh install.

mix ash_replicant.preflight --pipeline MyApp.Replicant.Pipeline



# Once deployed — everything above, plus checkpoint state, contract drift, and

# runtime readiness.

mix ash_replicant.doctor --pipeline MyApp.Replicant.Pipeline



# The same report as JSON, for a monitoring caller.

mix ash_replicant.doctor --pipeline MyApp.Replicant.Pipeline --format json

Both resolve the generated pipeline's own admitted start options, so you never restate configuration the application already carries. The same diagnosis is available in-process as AshReplicant.preflight/1 and AshReplicant.doctor/1, which take the option list AshReplicant.start_link/1 takes and return an AshReplicant.Doctor.Report. The Mix tasks verify the generated-pipeline marker from the BEAM export table before loading the named module, so an arbitrary --pipeline module cannot run @on_load or start_options/0 through a read-only command.

It performs no writes

Three independent legs, none of which trusts the other:

  1. Every source statement passes a fail-closed read-only admission — leading SELECT only, no statement separator, no write verb, no row lock, and no session-escaping function (set_config, dblink*).
  2. The probe connection is opened with default_transaction_read_only=on, so PostgreSQL itself refuses a write the admission missed. The live integration gate asserts exactly that, with the server's own read_only_sql_transaction SQLSTATE.
  3. The destination checkpoint is read through its :read action with authorize?: false and no lock — lock: :for_update is write intent and is never passed.

The commands never start a repo, a pipeline, or a service.

What the report distinguishes

Machine and operator output are both total functions of one canonical result, so they cannot disagree. Every class carries its own reason atom rather than a single "failed" bucket:

ClassReasons
Missing privilegesprivilege_replication_missing, privilege_select_missing, privilege_probe_missing
Unknown checkpoint statecheckpoint_state_unknown, checkpoint_state_key_unknown
Replica identitysource_replica_identity, judged independently of the rest of coverage
Retention horizonretention_extended → retention_at_risk → retention_lost
Contract driftcontract_drift reports the classifier's own reason (publication, relation_removed, stored_contract_invalid, …)
Version mismatchdependency_version_mismatch, dependency_missing, source_release_unsupported, source_release_untested

Reasons come from a closed vocabulary, and no connection option, publication name, source identity, slot name, watermark, or row value ever appears. A leg that could not be judged — an unreachable source, a repo that is not running — is reported skipped with the reason it could not be judged, never passed. If a connected server rejects or faults a catalog statement, reachability stays passed; the affected checks are source_probe_failed rather than falsely reported as unreachable.

Retention is the alert that must fire before recovery becomes impossible: retention_at_risk warns while the WAL is still there, retention_lost fails once it is not. A durable watermark whose slot has disappeared is already lost.

Exit codes

CodeMeaning
0Every check passed.
1At least one check failed.
2Warnings only.
3The invocation could not be diagnosed — missing, unknown, or unconfigured --pipeline, or an unknown flag.

3 is separate from 1 so a monitoring caller can tell "this deployment is unhealthy" from "you invoked me wrong".

Runtime readiness is node-local. The generation index is :persistent_term, so mix ash_replicant.doctor — its own OS process — always reports generation_absent. Call AshReplicant.doctor/1 from inside the running application (a remote console or a health endpoint) for the real answer.

Runtime status and lifecycle tombstones

AshReplicant.status/1 answers one question — what state is this sink's pipeline in? — with a closed five-value contract:

AshReplicant.status(MyApp.Replicant.Sink)

#=> :healthy | :catching_up | {:halted, reason} | {:misconfigured, reason} | :not_started

For every control-plane fault — source disconnect, owner death, checkpoint read/write faults, timeline changes, contract drift, census timeouts — the recovery-states matrix (docs/RECOVERY.md) documents the surfaced state, what the pipeline does on its own, and the operator action each one owes.

The answer is derived, never stored: it asks the live PipelineOwner for its own facts (census health, pipeline liveness), falls back to the node-local generation entry (a dead owner is the fault {:halted, :owner_lost} — mirroring has stopped, never "not started"), then to the tombstone legs. AshReplicant.Status.derive/2 exposes the six-state generation lifecycle underneath (:activating, :ready, :degraded, :halted, :stopped, :superseded) with its typed, value-free evidence.

Healthy is a strong claim: it requires a live owner and pipeline, an enabled census whose last run passed, and no in-flight snapshot. Owner liveness alone is insufficient — a pipeline whose census has not run yet (or is disabled) conservatively reports :catching_up, and so does one whose census is currently faulting below the halt budget.

Tombstones

When a generation ends, the party that knows the cause records a terminal tombstone — bounded (latest per slot), value-free (a closed reason atom, a class, a timestamp; never a row value, message prefix, or progress token). Every error leaving a sink callback records the scrubbed reason (Replicant halts the pipeline on any non-ok sink return, so that one boundary covers the halt funnels and the bind/session-identity/slot-origin error paths alike); the owner's census halt records the census reason; an operator stop records :operator_stopped. The tombstone has two legs: a node-local one (always writable) and a durable one on the checkpoint row (terminal_cause, terminal_class, terminal_at — added by mix ash.codegen + migrate), written only when the row already exists. Every admitted checkpoint write (bind — including the otherwise verify-only steady-state reconnect — and advance) clears the durable leg, so a stale cause never outlives the generation that superseded it.

Two documented edges: a halt while the destination is unreachable persists only the node-local leg (after a node restart the slot reports :not_started; the halt telemetry plus the :status, :tombstone_write_failed event is the durable record — the destination was down at the only moment the fact existed), and a host-tree shutdown writes no tombstone at all (no database writes during app teardown; a :stopped tombstone would map to the same public :not_started anyway). The halt window itself is closed: a status call made after any halt decision answers that halt's cause, never :healthy — activation clears the node-local leg before the generation entry exists, so a node-local tombstone under a live entry can only be that generation's own halt or stop decision, and it outranks the owner's facts. Replicant discards halt reasons at teardown, so a pipeline death nothing else explained records the generic {:halted, :pipeline_terminated} — over-alerting by design.

Observability and recovery horizons

Telemetry is value-free by construction: every event's metadata passes a typed allowlist (an off-allowlist key or an off-type value raises at the enforcement point), and the measurement key set is closed (count, change_count, duration, byte_size). emitted_event_names/0 lists the whole inventory; AshReplicant.Telemetry's moduledoc ships two executable examples — a dependency-free metrics reporter and an OpenTelemetry bridge whose mapping table is test-pinned complete against that inventory. The data-boundary mutation matrix carries one mutant per typed key, so a vacuous telemetry gate cannot ship.

Message-routed sinks additionally declare a recovery horizon — the outage/replay window every route's AshOnetime claim retention must cover:

use AshReplicant.Sink,

  ...,

  message_routes: [{"mail", MyApp.MailOutbox, :record}],

  recovery_horizon: {24, :hour}

Activation refuses :retention_below_recovery_horizon when any routed create's declared retention is shorter than the horizon: an in-window outage would expire a standalone message's only dedup while its WAL is still recoverable. The digest-key rotation window is witnessed durably: the checkpoint's authenticated digest_key_state envelope records the last-observed key set (under the orthogonal :ash_replicant, :horizon_provenance_keys family), and removing a key version within the retention horizon of the last observation that contained it halts :digest_key_horizon_violated instead of silently blocking future replays.

The alert table (what fires, what to do)

SignalMeaningOperator action
[:ash_replicant, :retention, :at_risk] (kind: :wal_unreserved or :wal_exhausted)The slot's WAL retention is being consumed while the pipeline runs — recovery is still possible but the window is shrinkingFind the lag (mix ash_replicant.doctor), resume or scale the sink before WAL is dropped
{:halted, :source_wal_lost} (census drift)The slot no longer retains the WAL the checkpoint needs — recovery through the slot is impossibleRestore the source from backup or re-snapshot; the slot cannot be resumed
activation refusal :retention_horizon_crossedThe pipeline was down longer than the shortest claim retention while the slot still had the WAL — re-delivery would re-execute standalone messagesReconcile the affected message routes (inspect for duplicates), then raise retention or restart with a fresh checkpoint decision
{:halted, :digest_key_horizon_violated} / :misconfigured, :digest_key_horizon_violatedA message-digest key version was removed while claims minted under it could still be re-deliveredRestore the removed key version to config, let the pipeline replay, retire it only past the retention horizon
:misconfigured, :retention_below_recovery_horizon (activation/doctor)A route's declared retention is shorter than the declared horizonRaise the route's retention({count, unit}) or lower recovery_horizon

The at-risk push fires from the census while the pipeline runs; while halted, nothing in the library watches the clock (the supervision contract owns no idle watcher) — run mix ash_replicant.doctor on the operator's own scheduler (cron, Kubernetes CronJob, or your alerting loop) as the periodic pull; its :slot_retention and :retention_horizon checks carry the same classes with per-check detail. That doctor cadence is the runbook: at-risk → resume before WAL drops; lost → restore; crossed → reconcile; violated → restore the key.

Upgrading to this surface: hosts run mix ash.codegen (the checkpoint gains the nullable digest_key_state column) and configure :ash_replicant, :horizon_provenance_keys (a {version, key} list, keys of at least 16 bytes) before starting a message-routed sink.

Strict source coverage

Every publication table is mapped, explicitly ignored (ignored_sources: ["public.audit_events"]), or the pipeline refuses to start; every delivered column is mapped or skipped, or delivery halts before writing. Column types are checked against the target at activation, and REPLICA IDENTITY FULL is enforced on tenant-scoped and non-PK-business-key SCD2 source tables. The preflight runs at activation (identity-verified, short-lived source connection) and the table-membership check re-runs at every reconnect — see ADR-0008 and usage-rules for the operator rules.

The PipelineOwner also runs that full coverage census periodically on a quiet stream. A table added to a publication, a type/RIF change, destination code or config drift, or a tampered checkpoint contract is therefore detected without waiting for reconnect or an affected row.

Effect-Once Semantics

Each transaction is applied in oneRepo.transaction:

  1. Skip any change whose commit_lsn <= checkpoint (watermark dedup).
  2. Apply each row change (upsert-by-PK, destroy, truncate per policy).
  3. Upsert the checkpoint (commit_lsn) in the same transaction.

On failure (schema change, multitenancy error, write fault), the entire transaction rolls back. The un-acked WAL re-streams on resume and dedups against the checkpoint.

Result for committed streaming transactions and provenance-backed V1/incremental snapshot retry: zero repeated host business effects and zero loss across replay, restart, and injected rollback faults. Snapshot chunks, provenance, exact incremental progress, and completion fences commit under the same source-bound checkpoint lock; unchanged retry rows perform bookkeeping only.

Append-log mode

A generated sink is exclusively a state mirror (the default: rows converge to the source's current state) or an append log (every change is recorded as an immutable event and nothing is ever modified or removed). The kind is declared on the sink, because Replicant exposes sink_kind/0 per sink rather than per resource; activation rejects a mixed resource set (ADR-0018).

defmodule MyApp.EventSink do

  use AshReplicant.Sink,

    repo: MyApp.Repo,

    domains: [MyApp.Events],

    checkpoint_resource: MyApp.ReplicantCheckpoint,

    slot_name: "shop_events",

    sink_kind: :append_log,

    # The ONE initial-state intent. `:go_forward` starts the log at the slot's

    # origin; `:snapshot` starts it from a full backfill. It must agree with the

    # `snapshot:` start option.

    initial_state: :go_forward,

    message_routes: [{"events", MyApp.Events.OrderEvent, :append}]

end

The append target is yours — this package generates no event table, no migration and no raw write path:

defmodule MyApp.Events.OrderEvent do

  use Ash.Resource,

    domain: MyApp.Events,

    data_layer: AshPostgres.DataLayer,

    extensions: [AshReplicant.Resource]



  postgres do

    table "order_events"

    repo MyApp.Repo

  end



  replicant do

    source_table "orders"

    append_log true

    # `:halt` (default) or `:append` — record the structural truncate event.

    on_truncate :append

  end



  attributes do

    uuid_primary_key :event_id



    # The five append-identity axes, plus the two structural labels and the

    # backfill attempt. Names are configurable (`append_commit_lsn_attribute`

    # and friends); these are the defaults.

    attribute :source_system_id, :string, allow_nil?: false, public?: true

    attribute :source_database, :string, allow_nil?: false, public?: true

    attribute :slot_name, :string, allow_nil?: false, public?: true

    attribute :commit_lsn, :integer, allow_nil?: false, public?: true

    attribute :ordinal, :integer, allow_nil?: false, public?: true

    attribute :operation, :string, allow_nil?: false, public?: true

    attribute :origin, :string, allow_nil?: false, public?: true

    attribute :snapshot_attempt, :binary, public?: true



    # The mapped payload — ordinary source columns, under the same skip,

    # sensitive and tenant rules a state mirror uses.

    attribute :id, :string, public?: true

    attribute :note, :string, public?: true

    # Destination-only logical-message payload. Content is binary-storage.

    attribute :message_prefix, :string, public?: true

    attribute :message_content, :binary, public?: true

  end



  identities do

    identity :append_identity, [

      :source_system_id, :source_database, :slot_name, :commit_lsn, :ordinal

    ]

  end



  actions do

    defaults [:read]



    # The IMMUTABLE create action — the only delivery path. Update, upsert and

    # destroy actions never are. Manual actions and arbitrary action/global

    # create changes are rejected because they can rewrite the identity after

    # input validation.

    create :append do

      accept [

        :source_system_id, :source_database, :slot_name, :commit_lsn, :ordinal,

        :operation, :origin, :snapshot_attempt, :id, :note,

        :message_prefix, :message_content

      ]

    end

  end

end

Back that identity with a real unique index (mix ash.codegen generates one from the identity) — it is the defensive database constraint behind append-once. On an attribute-multitenant append target the identity must also declare all_tenants? true, or Ash widens the upsert conflict target with the tenant discriminator; the compile verifier rejects the omission. All structural attributes except the stream-optional snapshot_attempt must be allow_nil?: false; PostgreSQL unique constraints do not collide NULLs.

Operation shapes.operation is "insert" | "update" | "delete" | "truncate" | "message" | "snapshot"; origin is "stream" | "snapshot". A delete appends the admitted old record, so the deleted payload survives in the log. Every append source table therefore requires REPLICA IDENTITY FULL; DEFAULT identity would supply only primary-key columns and silently truncate the delete event. A truncate is a structural event with no payload (and, being tenant-blind, is refused at compile time on a tenant-scoped target — use on_truncate :halt). Backfill rows carry the checkpoint-owned attempt id; they reuse none of the state-mirror provenance attributes. In append mode, each message_routes entry must target a non-tenant append resource's configured append action. A transactional message uses the transaction commit LSN and its shared ordinal; a standalone message uses its WAL LSN and ordinal zero. Prefix/content are written only to append_message_prefix_attribute / append_message_content_attribute (defaults :message_prefix / :message_content); they are destination-only and never treated as source-table columns. State-mirror message routes keep the ADR-0015 AshOnetime contract unchanged.

Append-once. The append identity is exactly (source system, database, slot, commit LSN, ordinal), and delivery upserts against it with an empty upsert_fields — a no-op conflict clause. A lawfully re-delivered WAL position (a crash redo, a snapshot re-run) appends once and never rewrites the stored event. Effect-once itself is unchanged: the append and the checkpoint commit in one locked destination transaction.

The origin floor. A :go_forward sink records the slot origin it first started from on the checkpoint's origin_floor, once. No completeness claim covers data below it. Later reconnect origins are resume facts; a slot recreated under an existing floor halts :append_origin_gap, and an appended event above the durable checkpoint halts :append_frontier_divergent. Replicant does not filtered-WAL idle-advance an append sink, so a reused origin ahead of the durable checkpoint is an unambiguous :append_origin_gap. Publish a normal heartbeat transaction on a quiet append publication to advance the checkpoint and release retained WAL.

Boundaries. An append sink cannot run incremental snapshots (that mode requires snapshot_provenance on every mapped resource, which an append target may not declare), and a strategy :context append target is refused on a go-forward sink. Separate mirror and append pipelines may use separate slots and checkpoints; one slot is not claimed to serve both kinds.

Destination transaction boundary

Every admitted destination resource uses the sink's literal AshPostgres Repo and the same effective dynamic Repo. The activation manifest starts from the checkpoint and mapped read/create/destroy/SCD2-close actions, recursively follows framework relationships and declared custom actions, and rejects a missing action, foreign or dynamic Repo, non-Postgres data layer, recursive cycle, or touches_resources mismatch before delivery.

Generated sink callbacks are final and invoke the admitted implementation directly; hosts cannot override an effect hook to bypass apply/checkpoint. A SetContext change or preparation that replaces :data_layer is rejected (a dynamic/MFA context is admissible only when its module declares DestinationParticipant), as is any AshOnetime cache other than the effect-free AshOnetime.Cache.None.

Arbitrary changes, validations, preparations, manual actions, callbacks, custom types, and tenant resolvers must implement AshReplicant.DestinationParticipant. Return :no_database or literal Ash resource/action references. Declarations are trusted metadata; they do not prove an arbitrary Elixir body. A declaration cannot make raw SQL, another Repo, asynchronous work, or an external effect part of the atomic guarantee.

defmodule MyApp.ReplicantReceiptParticipant do

  @behaviour AshReplicant.DestinationParticipant



  alias AshReplicant.DestinationParticipant.{ActionRef, Context, ReplayIdentity}



  @impl AshReplicant.DestinationParticipant

  def destination_participants(_opts, %Context{}) do

    {:ok,

     {:actions,

      [

        %ActionRef{

          resource: MyApp.MirrorReceipt,

          action: :record,

          replay_identity: %ReplayIdentity{

            participant: :mirror_receipt,

            components: [

              :source_system_identifier,

              :source_database,

              :slot_name,

              :commit_lsn,

              :ordinal,

              :participant

            ]

          }

        }

      ]}}

  end

end

The containing Ash action's touches_resources must exactly match the resources discovered from its providers. When the declared auxiliary action needs a replay guard, it may use only local AshOnetime idempotency committed with the action in the admitted Repo. It must take the private, non-null operation_key produced by AshReplicant.DestinationParticipant.operation_key/2 and use the exact versioned participant scope and replay identity shown above. AshOnetime one-time nonces are rejected for WAL replay. Independent commits and external effects are rejected too.

A Replicant v1 retry and incremental resume are physically effect-once for resources declaring snapshot_provenance true: fingerprints suppress repeated host actions, and incremental progress commits atomically with each bounded chunk. Message (C1), sink-owned batch delivery (C2), and incremental progress (C3) are live — batch_delivery opts a pipeline into handle_batch/1, one destination transaction and one watermark write per flushed batch (ADR-0016). Append-log delivery is not exported yet and must compose with this same boundary. See ADR-0006.

Multitenancy & Classification

Sensitive Data

Sensitive source columns must map to one of:

  1. AshCloak-encrypted — the before_action hook fires on upsert.
  2. Binary storage — user-managed encryption (store and load encrypted values).
  3. Skipped — excluded from the mirror (listed in skip).

The verifier runs at compile time and rejects a resource if a sensitive column violates one of these rules.

Development

asdf install

scripts/with-release-runtime.sh scripts/assert-runtime-version.sh

scripts/with-release-runtime.sh mix deps.get

env -u ASH_REPLICANT_TEST_URL \

  scripts/with-release-runtime.sh scripts/run-structural-tests.sh \

    --allow-excluded --exclude integration



export ASH_REPLICANT_TEST_URL="postgres://postgres@localhost:5599/postgres" # example — point at YOUR logical-replication Postgres (host/port are machine-local; the database name is forced anyway)

MIX_ENV=test scripts/with-release-runtime.sh mix ecto.create

MIX_ENV=test scripts/with-release-runtime.sh mix ecto.migrate

scripts/with-release-runtime.sh scripts/run-structural-tests.sh --include integration

scripts/with-release-runtime.sh scripts/run-structural-tests.sh \

  test/integration --include integration



scripts/with-release-runtime.sh scripts/test-migration-drift-gate.sh

scripts/with-release-runtime.sh scripts/test-release-checkers.sh

scripts/with-release-runtime.sh scripts/test-release-contract.sh



scripts/with-release-runtime.sh mix format --check-formatted

scripts/with-release-runtime.sh mix compile --warnings-as-errors

scripts/with-release-runtime.sh mix credo --strict

scripts/with-release-runtime.sh mix deps.audit

scripts/with-release-runtime.sh mix dialyzer

scripts/with-release-runtime.sh mix docs --warnings-as-errors

scripts/with-release-runtime.sh mix hex.build

All gates pass before commit. Update CHANGELOG.md under [Unreleased].

License

MIT — see LICENSE.