External effects and recovery

Copy Markdown View Source
Run in Livebook 
Mix.install(
  [
    {:ash_onetime, "~> 1.5.0"},
    {:ash_postgres, "~> 2.13"}
  ],
  consolidate_protocols: false
)

A result can outlive the caller's transaction

An external effect can commit while the transaction recording its response rolls back. Retrying blindly would duplicate it. ash_onetime reserves a durable operation key before execution; a retry uses that same key to recover the accepted result.

This notebook's effect is a durable PostgreSQL receipt ledger. A separate Repo and connection pool commit the receipt independently of the protected action's transaction. You will observe fresh execution, stored replay, recovery after a real rollback, and a concurrent retry refused while the peer is blocked on a real database lock.

This is an executable database adapter example. An HTTP or payment adapter needs its own peer's documented idempotency and authoritative recovery API; see the external-effects guide.

Requires Elixir 1.20+, Ash 3.34.3+, and PostgreSQL 18 at DATABASE_URL. Use a development database where you can create and remove schemas. The final cell removes this run's admission objects, receipt ledger, and invocation log.

Connect the action and ledger independently

database_url = System.fetch_env!("DATABASE_URL")
Logger.configure(level: :warning)
:ok = Application.put_env(:ash, :disable_async?, true)
:ok = Application.put_env(:ash, :default_string_length_count, :codepoints)

for repo <- [AshOnetimeDemo.Repo, AshOnetimeDemo.PeerRepo] do
  :ok = Application.put_env(:ash_onetime_demo, repo, url: database_url, pool_size: 10)
end

defmodule AshOnetimeDemo.Repo do
  use AshPostgres.Repo, otp_app: :ash_onetime_demo
  @impl AshPostgres.Repo
  def installed_extensions, do: ["ash-functions"]
  @impl AshPostgres.Repo
  def min_pg_version, do: %Version{major: 18, minor: 0, patch: 0}
end

defmodule AshOnetimeDemo.PeerRepo do
  use Ecto.Repo, otp_app: :ash_onetime_demo, adapter: Ecto.Adapters.Postgres
end

{:ok, _} = AshOnetimeDemo.Repo.start_link()
{:ok, _} = AshOnetimeDemo.PeerRepo.start_link()

Implement the ledger's execute and recover operations

The operation key is the claim UUID supplied by the library. The ledger's primary key admits one receipt per operation; its append-only invocation log records each callback. Each log insert commits before the receipt transaction, so a failed receipt write cannot erase its invocation. The log exposes repeated executions even if receipt deduplication hides their effect. recover/3 returns :absent only after a successful authoritative lookup finds no receipt. A failed lookup returns :unknown.

defmodule AshOnetimeDemo.PeerAdapter do
  @behaviour AshOnetime.ExternalEffect
  alias AshOnetimeDemo.PeerRepo
  alias Ecto.Adapters.SQL

  @impl true
  def execute(operation_key, input, _context) do
    schema = schema!(input)
    value = Ash.ActionInput.get_argument(input, :value)

    log_call!(schema, operation_key, "execute")

    PeerRepo.transaction(fn ->
      %{rows: [[stored]]} = SQL.query!(PeerRepo, """
      INSERT INTO "#{schema}".external_receipts (operation_key, value)
      VALUES ($1::text::uuid, $2)
      ON CONFLICT (operation_key) DO UPDATE SET operation_key = EXCLUDED.operation_key
      RETURNING value
      """, [operation_key, value], timeout: 10_000)
      stored
    end, timeout: 15_000)
  rescue
    _ -> {:error, :outcome_unknown}
  end

  @impl true
  def recover(operation_key, input, _context) do
    schema = schema!(input)
    log_call!(schema, operation_key, "recover")

    case SQL.query!(PeerRepo, """
         SELECT value FROM "#{schema}".external_receipts
         WHERE operation_key = $1::text::uuid
         """, [operation_key], timeout: 5_000).rows do
      [[value]] -> {:ok, value}
      [] -> :absent
    end
  rescue
    _ -> :unknown
  end

  defp log_call!(schema, operation_key, kind) do
    SQL.query!(PeerRepo, """
    INSERT INTO "#{schema}".external_invocations (operation_key, kind)
    VALUES ($1::text::uuid, $2)
    """, [operation_key, kind], timeout: 5_000)
  end

  defp schema!(input) do
    schema = input.tenant
    true = is_binary(schema) and Regex.match?(~r/^ash_onetime_demo_[a-z0-9_]+$/, schema)
    schema
  end
end

defmodule AshOnetimeDemo.ChargeCodec do
  @behaviour AshOnetime.Codec
  @impl true
  def format_tag, do: "charge-v1"
  @impl true
  def encode(value, _contract, _opts), do: {:ok, format_tag(), :erlang.term_to_binary(value)}
  @impl true
  def decode("charge-v1", payload, _contract, _opts), do: {:ok, :erlang.binary_to_term(payload, [:safe])}
end

defmodule AshOnetimeDemo.ChargeClassifier do
  def classify(value, _context), do: {:store, value}
end

defmodule AshOnetimeDemo.RedeemRun do
  use Ash.Resource.Actions.Implementation
  @impl true
  def run(input, _opts, _context), do: AshOnetime.ExternalEffect.result(input)
end

Protect the action

The action returns the accepted ledger result through ExternalEffect.result/1. It does not reconstruct a result from request input.

defmodule AshOnetimeDemo.Domain do
  use Ash.Domain, validate_config_inclusion?: false
  resources do
    resource AshOnetimeDemo.ExternalRedeem
  end
end

defmodule AshOnetimeDemo.ExternalRedeem do
  use Ash.Resource,
    domain: AshOnetimeDemo.Domain,
    data_layer: AshPostgres.DataLayer,
    extensions: [AshOnetime.Resource]

  postgres do
    table "demo_external_redeem"
    repo AshOnetimeDemo.Repo
  end
  multitenancy do
    strategy :context
  end
  attributes do
    uuid_primary_key :id
  end
  actions do
    defaults [:read]
    action :redeem, :integer do
      transaction? true
      argument :value, :integer, allow_nil?: false
      argument :request_key, :string, allow_nil?: false
      run {AshOnetimeDemo.RedeemRun, []}
    end
  end
  onetime do
    protect :redeem do
      strategy :idempotency
      scope([{:static, "external"}])
      key({:client, :request_key})
      fingerprint(arguments: [:value])
      response(AshOnetimeDemo.ChargeCodec, fields: [], classify: AshOnetimeDemo.ChargeClassifier)
      retention({1, :hour})
      external_effect(AshOnetimeDemo.PeerAdapter)
    end
  end
end

:ok

Create this run's store and ledger

alias AshOnetimeDemo.{Repo, PeerRepo}
alias Ecto.Adapters.SQL
unique = Ecto.UUID.generate() |> String.replace("-", "")
schema = "ash_onetime_demo_external_#{unique}"
temp = Path.join(System.tmp_dir!(), schema)
File.mkdir_p!(temp)
schema
source = Mix.Tasks.AshOnetime.Gen.Migrations.render(Repo,
  hash_partitions: nil,
  partition_start: Date.utc_today() |> Date.beginning_of_month())
timestamp = "20260101000000"
path = Path.join(temp, "#{timestamp}_install_ash_onetime.exs")
File.write!(path, source)
[{module, _}] = Code.compile_file(path)
version = String.to_integer(timestamp)
SQL.query!(Repo, ~s{CREATE SCHEMA "#{schema}"}, [])
:ok = Ecto.Migrator.up(Repo, version, module, prefix: schema, log: false)
SQL.query!(Repo, ~s{CREATE TABLE "#{schema}".demo_external_redeem (id uuid PRIMARY KEY)}, [])
SQL.query!(PeerRepo, """
CREATE TABLE "#{schema}".external_receipts (operation_key uuid PRIMARY KEY, value bigint NOT NULL CHECK (value >= 0))
""", [])
SQL.query!(PeerRepo, """
CREATE TABLE "#{schema}".external_invocations (
  id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  operation_key uuid NOT NULL,
  kind text NOT NULL CHECK (kind IN ('execute', 'recover'))
)
""", [])

run = fn key, value ->
  AshOnetimeDemo.ExternalRedeem
  |> Ash.ActionInput.for_action(:redeem, %{value: value, request_key: key})
  |> Ash.ActionInput.set_tenant(schema)
  |> Ash.run_action()
end
call_count = fn kind ->
  [[count]] = SQL.query!(PeerRepo,
    ~s{SELECT count(*) FROM "#{schema}".external_invocations WHERE kind = $1}, [kind]).rows
  count
end
:ok

Execute once, then replay the stored result

The first call commits one receipt. A successful retry replays the locally stored result: it needs neither a second execute nor recovery lookup. Primitive-return actions do not carry a record replay marker, so we observe the actual callback log instead.

{:ok, 42} = run.("ext-1", 42)
{:ok, 42} = run.("ext-1", 42)
1 = call_count.("execute")
0 = call_count.("recover")
%{value: 42, execute_calls: 1, recover_calls: 0}
%{execute_calls: 1, recover_calls: 0, value: 42}

Roll back the caller after the ledger commits

The caller rolls back its actual transaction after the protected action returns. The independently committed receipt survives; the admission claim survives as processing, without a completed local replay response.

{:error, :caller_abort} = Repo.transaction(fn ->
  {:ok, 99} = run.("ext-recovery-1", 99)
  Repo.rollback(:caller_abort)
end)
[[operation_key]] = SQL.query!(PeerRepo,
  ~s{SELECT operation_key::text FROM "#{schema}".external_receipts WHERE value = 99}, []).rows
[["processing"]] = SQL.query!(Repo,
  ~s{SELECT state FROM "#{schema}".ash_onetime_idempotency_claims WHERE id = $1::text::uuid},
  [operation_key]).rows
%{caller_rolled_back: true, durable_peer_value: 99, claim_state: "processing"}
%{caller_rolled_back: true, claim_state: "processing", durable_peer_value: 99}

Recover the result under the same operation key

Retrying this incomplete claim invokes recover/3. The ledger returns the existing receipt and the action completes, without another execution. The append-only callback log proves both decisions for this exact operation key.

{:ok, 99} = run.("ext-recovery-1", 99)
[[1, 1]] = SQL.query!(PeerRepo, """
SELECT count(*) FILTER (WHERE kind = 'execute'), count(*) FILTER (WHERE kind = 'recover')
FROM "#{schema}".external_invocations WHERE operation_key = $1::text::uuid
""", [operation_key]).rows
[[1]] = SQL.query!(PeerRepo,
  ~s{SELECT count(*) FROM "#{schema}".external_receipts WHERE operation_key = $1::text::uuid},
  [operation_key]).rows
%{value: 99, execute_calls: 1, recover_calls: 1, receipts: 1}
%{execute_calls: 1, receipts: 1, recover_calls: 1, value: 99}

Observe a real blocked peer, then retry concurrently

A separate PostgreSQL session takes a table lock on the ledger. The original's receipt insert blocks on that real lock. We confirm the wait in pg_locks before issuing a retry: the retry cannot acquire the pre-peer claim lock and returns :request_in_progress within external_lock_timeout_ms (default 2000).

The lock holder releases on completion or after a bounded wait. It affects only this notebook's unique schema.

parent = self()
locker = Task.async(fn ->
  PeerRepo.transaction(fn ->
    SQL.query!(PeerRepo, ~s{LOCK TABLE "#{schema}".external_receipts IN ACCESS EXCLUSIVE MODE}, [])
    send(parent, :ledger_locked)
    receive do
      :release -> :ok
    after
      15_000 -> PeerRepo.rollback(:lock_holder_deadline)
    end
  end, timeout: 20_000)
end)
receive do
  :ledger_locked -> :ok
after
  5_000 -> raise "ledger lock was not acquired"
end
original = Task.async(fn -> run.("ext-lock-1", 7) end)

retry_code = try do
  blocked = Enum.reduce_while(1..200, false, fn _, _ ->
    [[count]] = SQL.query!(Repo, """
    SELECT count(*) FROM pg_locks l JOIN pg_stat_activity a ON a.pid = l.pid
    WHERE l.relation = $1::text::regclass AND NOT l.granted
      AND a.wait_event_type = 'Lock' AND a.query LIKE 'INSERT INTO%external_receipts%'
    """, [~s{"#{schema}".external_receipts}]).rows
    if count == 1 do
      {:halt, true}
    else
      Process.sleep(20)
      {:cont, false}
    end
  end)
  true = blocked
  {:error, refusal} = run.("ext-lock-1", 7)
  :request_in_progress = AshOnetime.Error.code(refusal)
after
  send(locker.pid, :release)
end

{:ok, :ok} = Task.await(locker, 10_000)
{:ok, 7} = Task.await(original, 10_000)
3 = call_count.("execute")
1 = call_count.("recover")
%{concurrent_retry: retry_code, total_execute_calls: 3}
%{concurrent_retry: :request_in_progress, total_execute_calls: 3}

The concurrent retry never called the peer. Once the original completes, a later retry can replay its stored result. For an unreachable peer, :unknown means the library cannot authorize another execution. Adapters must bound calls and preserve that distinction. See operations for abandoned-claim reconciliation and the reaper's boundary.

Observe a failed receipt write

The ledger rejects negative values with a real PostgreSQL constraint. After the failed execute, an authoritative recovery lookup confirms that no receipt exists; the action returns :external_effect_unavailable. Its separately committed invocation remains visible:

before_failed_execute = call_count.("execute")
{:error, failed_receipt} = run.("negative-receipt", -1)
:external_effect_unavailable = AshOnetime.Error.code(failed_receipt)
true = call_count.("execute") == before_failed_execute + 1
[[0]] = SQL.query!(PeerRepo,
  ~s{SELECT count(*) FROM "#{schema}".external_receipts WHERE value = -1}, []).rows
%{failed_execute_logged: true, receipt_count: 0}
%{failed_execute_logged: true, receipt_count: 0}

Clean up

:ok = Ecto.Migrator.down(Repo, version, module, prefix: schema, log: false)
SQL.query!(Repo, ~s{DROP SCHEMA "#{schema}" CASCADE}, [])
File.rm_rf!(temp)
:ok

Continue