<!--
livebook:{"persist_outputs":true}
-->

# External effects and recovery

```elixir
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](https://hexdocs.pm/ash_onetime/external-effects.html).

> 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

```elixir
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`.

```elixir
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.

```elixir
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

```elixir
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
```

```elixir
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.

```elixir
{: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}
```

<!-- livebook:{"output":true} -->

```
%{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.

```elixir
{: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"}
```

<!-- livebook:{"output":true} -->

```
%{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.

```elixir
{: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}
```

<!-- livebook:{"output":true} -->

```
%{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.

```elixir
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}
```

<!-- livebook:{"output":true} -->

```
%{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](https://hexdocs.pm/ash_onetime/operations.html) 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:

```elixir
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}
```

<!-- livebook:{"output":true} -->

```
%{failed_execute_logged: true, receipt_count: 0}
```

## Clean up

```elixir
: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

- [Idempotency](idempotency.livemd) — local effects and exact replay.
- [One-time nonces](nonces.livemd) — proof verification and collision rejection.
- [External effects](https://hexdocs.pm/ash_onetime/external-effects.html) — adapter and peer requirements.
- [Transaction-owned admission](https://hexdocs.pm/ash_onetime/transaction-owned-admission.html) — compose admission in your transaction.