Packages
ash
2.21.7
3.29.3
3.29.2
3.29.1
3.29.0
3.28.0
3.27.8
3.27.7
3.27.6
3.27.5
3.27.4
3.27.3
3.27.2
3.27.1
3.27.0
3.26.0
3.25.2
3.25.1
3.25.0
retired
3.24.7
3.24.6
3.24.5
3.24.4
3.24.3
3.24.2
3.24.1
3.24.0
3.23.1
3.23.0
3.22.2
3.22.1
3.22.0
3.21.3
3.21.2
3.21.1
3.21.0
3.20.0
3.19.3
3.19.2
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.0
3.15.0
3.14.1
3.14.0
retired
3.13.2
3.13.1
3.13.0
3.12.0
3.11.3
3.11.2
3.11.1
3.11.0
3.10.1
3.10.0
3.9.0
3.8.0
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
retired
3.6.3
retired
3.6.2
3.6.1
3.6.0
3.5.43
3.5.42
3.5.41
3.5.40
3.5.39
3.5.38
3.5.37
3.5.36
3.5.35
3.5.34
3.5.33
3.5.32
3.5.31
3.5.30
3.5.29
3.5.28
3.5.27
3.5.26
3.5.25
3.5.24
3.5.23
3.5.22
3.5.21
3.5.20
3.5.19
3.5.18
3.5.17
3.5.16
3.5.15
3.5.14
3.5.13
3.5.12
3.5.11
3.5.10
3.5.9
3.5.8
3.5.7
3.5.6
3.5.5
3.5.4
3.5.3
3.5.2
3.5.1
3.5.0
3.4.74
retired
3.4.73
3.4.72
3.4.71
3.4.70
3.4.69
3.4.68
3.4.67
3.4.66
3.4.65
3.4.64
3.4.63
3.4.62
3.4.61
3.4.60
3.4.59
3.4.58
3.4.57
3.4.56
3.4.55
3.4.54
3.4.53
3.4.52
3.4.51
3.4.50
3.4.49
3.4.48
3.4.47
3.4.46
3.4.45
3.4.44
3.4.43
3.4.42
3.4.41
3.4.40
3.4.39
3.4.38
3.4.37
3.4.36
3.4.35
3.4.34
3.4.33
3.4.32
3.4.31
3.4.30
3.4.29
3.4.28
3.4.27
3.4.26
3.4.25
3.4.24
3.4.23
3.4.22
3.4.21
3.4.20
3.4.19
3.4.18
3.4.17
3.4.16
3.4.15
3.4.14
3.4.13
3.4.12
3.4.11
3.4.10
3.4.9
3.4.8
3.4.7
3.4.6
3.4.5
3.4.4
3.4.3
3.4.2
3.4.1
3.4.0
3.3.3
3.3.2
3.3.1
3.3.0
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.16
3.0.15
3.0.14
3.0.13
3.0.12
3.0.11
3.0.10
3.0.9
3.0.8
3.0.7
3.0.6
3.0.5
3.0.4
3.0.3
3.0.2
3.0.1
3.0.0
3.0.0-rc.46
3.0.0-rc.45
3.0.0-rc.44
3.0.0-rc.43
3.0.0-rc.42
3.0.0-rc.41
3.0.0-rc.40
3.0.0-rc.39
3.0.0-rc.38
3.0.0-rc.37
3.0.0-rc.36
3.0.0-rc.35
3.0.0-rc.34
3.0.0-rc.33
3.0.0-rc.32
3.0.0-rc.31
3.0.0-rc.29
3.0.0-rc.28
3.0.0-rc.27
3.0.0-rc.26
3.0.0-rc.25
3.0.0-rc.24
3.0.0-rc.23
3.0.0-rc.22
3.0.0-rc.21
3.0.0-rc.20
3.0.0-rc.19
3.0.0-rc.18
3.0.0-rc.17
3.0.0-rc.16
3.0.0-rc.15
3.0.0-rc.14
3.0.0-rc.13
3.0.0-rc.12
3.0.0-rc.11
3.0.0-rc.10
3.0.0-rc.9
3.0.0-rc.8
3.0.0-rc.7
3.0.0-rc.6
3.0.0-rc.5
3.0.0-rc.4
3.0.0-rc.3
3.0.0-rc.1
3.0.0-rc.0
2.21.15
2.21.14
2.21.13
2.21.12
2.21.11
2.21.10
2.21.9
2.21.8
2.21.7
2.21.6
2.21.5
2.21.4
2.21.3
2.21.2
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.14
2.19.13
2.19.12
2.19.11
2.19.10
2.19.9
2.19.8
2.19.7
2.19.6
2.19.5
2.19.4
2.19.3
retired
2.19.2
retired
2.19.1
retired
2.19.0
retired
2.18.2
2.18.1
2.18.0
2.17.24
2.17.23
2.17.22
2.17.21
2.17.20
2.17.19
2.17.18
2.17.17
2.17.16
2.17.15
2.17.14
2.17.13
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.1
2.16.0
2.15.20
2.15.19
2.15.18
2.15.17
2.15.16
2.15.15
2.15.14
2.15.13
2.15.12
2.15.11
2.15.10
2.15.9
2.15.8
2.15.7
2.15.6
2.15.5
2.15.4
2.15.2
2.15.1
2.15.0
2.14.21
2.14.20
2.14.19
2.14.18
2.14.17
2.14.16
2.14.15
2.14.14
2.14.13
2.14.12
2.14.11
2.14.10
2.14.9
2.14.8
2.14.7
2.14.6
2.14.5
2.14.4
2.14.3
2.14.2
2.14.1
2.14.0
2.13.4
retired
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.11
2.11.10
2.11.9
2.11.8
2.11.7
2.11.6
2.11.5
2.11.4
2.11.3
2.11.2
2.11.1
2.11.0
2.11.0-rc.3
2.11.0-rc.2
2.11.0-rc.1
2.11.0-rc.0
2.10.2
2.10.1
2.10.0
2.9.29
2.9.28
2.9.27
2.9.26
2.9.25
2.9.24
2.9.23
2.9.22
2.9.21
2.9.20
2.9.19
2.9.18
2.9.17
2.9.16
2.9.15
2.9.14
2.9.13
2.9.12
2.9.11
2.9.10
2.9.9
2.9.8
2.9.7
2.9.6
2.9.5
2.9.4
2.9.3
2.9.2
2.9.1
2.9.0
2.8.1
2.8.0
2.7.1
2.7.0
2.6.31
2.6.30
2.6.29
2.6.28
2.6.27
2.6.26
2.6.25
2.6.24
2.6.23
2.6.22
2.6.21
2.6.20
2.6.19
2.6.18
2.6.17
2.6.16
2.6.15
2.6.14
2.6.13
2.6.11
2.6.10
2.6.9
2.6.8
2.6.7
2.6.6
2.6.5
2.6.4
2.6.3
2.6.2
2.6.1
2.6.0
2.5.16
2.5.15
2.5.14
2.5.13
2.5.12
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.5.0-rc.6
2.5.0-rc.5
2.5.0-rc.4
2.5.0-rc.3
2.5.0-rc.2
2.5.0-rc.1
2.5.0-rc.0
2.4.30
2.4.29
2.4.28
2.4.27
2.4.26
2.4.25
2.4.24
2.4.23
2.4.22
2.4.21
2.4.20
2.4.19
2.4.18
2.4.17
2.4.16
2.4.15
2.4.14
2.4.13
2.4.12
2.4.11
2.4.10
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.15
2.0.0-rc.14
2.0.0-rc.13
2.0.0-rc.12
2.0.0-rc.11
2.0.0-rc.10
2.0.0-rc.9
2.0.0-rc.8
2.0.0-rc.7
2.0.0-rc.6
2.0.0-rc.5
2.0.0-rc.4
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
2.0.0-pre.8
2.0.0-pre.7
2.0.0-pre.6
2.0.0-pre.5
2.0.0-pre.4
2.0.0-pre.3
2.0.0-pre.2
2.0.0-pre.1
2.0.0-pre.0
1.53.3
1.53.2
1.53.0
1.52.0-rc.22
1.52.0-rc.21
1.52.0-rc.20
1.52.0-rc.19
1.52.0-rc.18
1.52.0-rc.17
1.52.0-rc.16
1.52.0-rc.15
1.52.0-rc.14
1.52.0-rc.13
1.52.0-rc.12
1.52.0-rc.11
1.52.0-rc.10
1.52.0-rc.9
1.52.0-rc.8
1.52.0-rc.7
1.52.0-rc.6
1.52.0-rc.5
1.52.0-rc.4
1.52.0-rc.3
1.52.0-rc.2
1.52.0-rc.1
1.52.0-rc.0
1.51.2
1.51.1
retired
1.51.0
1.50.21
1.50.20
1.50.19
1.50.18
1.50.17
1.50.16
1.50.15
1.50.14
1.50.13
1.50.12
1.50.11
1.50.10
1.50.9
1.50.8
1.50.7
1.50.6
1.50.5
1.50.4
1.50.3
1.50.2
1.50.1
1.50.0
1.49.0
1.48.0-rc.30
1.48.0-rc.29
1.48.0-rc.28
1.48.0-rc.27
1.48.0-rc.26
1.48.0-rc.25
1.48.0-rc.24
1.48.0-rc.23
1.48.0-rc.22
1.48.0-rc.21
1.48.0-rc.20
1.48.0-rc.19
1.48.0-rc.18
1.48.0-rc.17
1.48.0-rc.16
1.48.0-rc.15
1.48.0-rc.14
1.48.0-rc.13
1.48.0-rc.12
1.48.0-rc.11
1.48.0-rc.10
1.48.0-rc.9
1.48.0-rc.8
1.48.0-rc.7
1.48.0-rc.6
1.48.0-rc.5
1.48.0-rc.4
1.48.0-rc.3
1.48.0-rc.2
1.48.0-rc.1
1.48.0-rc.0
1.47.12
1.47.11
1.47.10
1.47.9
1.47.8
1.47.7
1.47.6
1.47.5
1.47.4
1.47.3
1.47.2
1.47.1
1.47.0
1.46.13
1.46.12
1.46.11
1.46.10
1.46.9
1.46.8
1.46.7
1.46.6
1.46.5
1.46.4
1.46.3
1.46.2
1.46.1
1.46.0
1.45.0-rc9
1.45.0-rc8
1.45.0-rc7
1.45.0-rc6
1.45.0-rc5
1.45.0-rc4
1.45.0-rc3
1.45.0-rc20
1.45.0-rc2
1.45.0-rc19
1.45.0-rc18
1.45.0-rc17
1.45.0-rc16
1.45.0-rc15
1.45.0-rc14
1.45.0-rc13
1.45.0-rc12
1.45.0-rc11
1.45.0-rc10
1.45.0-rc1
1.45.0-rc0
1.44.13
1.44.12
1.44.11
1.44.10
1.44.9
1.44.8
1.44.7
1.44.6
1.44.5
1.44.4
1.44.3
1.44.2
1.44.1
1.44.0
1.43.12
1.43.11
1.43.10
1.43.9
1.43.8
1.43.7
1.43.6
1.43.5
1.43.4
1.43.3
1.43.2
1.43.1
1.43.0
1.42.0
1.41.12
1.41.11
1.41.10
1.41.9
1.41.8
1.41.7
1.41.6
1.41.5
1.41.4
1.41.3
1.41.2
1.41.1
1.41.0
1.40.0
1.39.7
1.39.6
1.39.5
1.39.4
1.39.3
1.39.2
1.39.1
1.39.0
1.38.0
1.37.2
1.37.1
1.37.0
1.36.22
1.36.21
1.36.19
1.36.18
1.36.17
1.36.16
1.36.15
1.36.14
1.36.13
1.36.12
1.36.11
1.36.10
1.36.9
1.36.8
1.36.7
1.36.6
1.36.5
1.36.4
1.36.3
1.36.2
1.36.0
1.35.1
1.35.0
1.34.9
1.34.8
1.34.7
1.34.6
1.34.5
1.34.4
1.34.3
1.34.2
1.34.1
1.34.0
1.33.0
1.32.2
1.32.1
1.32.0
1.31.1
1.31.0
1.30.2
1.30.1
1.29.0-rc1
1.29.0-rc0
1.28.1
1.28.0
1.27.1
1.27.0
1.26.13
1.26.12
1.26.11
1.26.10
1.26.9
1.26.8
1.26.7
1.26.6
1.26.5
1.26.4
1.26.2
1.26.1
1.26.0
1.25.8
1.25.7
1.25.6
1.25.5
1.25.4
1.25.3
1.25.2
1.25.1
1.25.0
1.24.2
1.24.1
1.24.0
1.23.3
1.23.2
1.23.1
1.23.0
1.22.1
1.22.0
1.20.1
1.20.0
1.19.1
1.19.0
1.18.1
1.18.0
1.17.1
1.17.0
1.16.2
1.15.1
1.15.0
1.14.0
1.13.4
1.13.3
1.13.2
1.13.1
1.13.0
1.12.0
1.11.1
1.11.0
1.10.0
1.9.0
1.8.0
1.7.0
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.1
1.4.0
1.3.1
1.3.0
1.2.1
1.2.0
1.1.3
1.1.2
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
0.13.1
0.13.0
0.12.0
0.10.0
0.9.1
0.9.0
0.8.0
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.0
0.3.0
0.2.0
0.1.9
0.1.8
0.1.3
0.1.1
0.1.0
A declarative, extensible framework for building Elixir applications.
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/ash/engine/engine.ex
defmodule Ash.Engine do
@moduledoc false
defstruct [
:ref,
:id,
:resolved_fields,
:authorize?,
:actor,
:verbose?,
:async?,
:concurrency_limit,
# :continue - Go until the end, gathering all errors
# :stop - Stop at the first error
failure_mode: :stop,
return_notifications?: false,
opts: [],
requests: [],
data: %{},
dependencies_waiting_on_request: MapSet.new(),
unsent_dependencies: [],
dependencies_seen: MapSet.new(),
dependencies: %{},
reverse_dependencies: %{},
resource_notifications: [],
tasks: [],
errors: [],
notifications: MapSet.new(),
pending_tasks: []
]
alias Ash.Engine.Request
require Ash.Tracer
require Logger
def run(requests, opts \\ []) do
resources =
opts[:resource]
|> List.wrap()
|> Enum.concat(Enum.flat_map(requests, &(&1.touches_resources || [])))
|> Enum.concat(Enum.flat_map(requests, &List.wrap(&1.resource)))
|> Enum.concat(opts[:touches_resources] || [])
|> Enum.reject(&is_nil/1)
|> Enum.filter(&Ash.DataLayer.data_layer_can?(&1, :transact))
cond do
opts[:transaction?] && Enum.empty?(resources) ->
raise "Engine invoked with `transaction?: true` but no resource, so no transaction could be started."
opts[:transaction?] ->
resources
|> Enum.reject(&Ash.DataLayer.in_transaction?/1)
|> case do
[] ->
do_run(requests, opts)
resources ->
notify? =
if Process.get(:ash_started_transaction?) do
false
else
Process.put(:ash_started_transaction?, true)
true
end
try do
Ash.DataLayer.transaction(
resources,
fn ->
case do_run(requests, opts) do
{:ok, result} ->
result
{:error, error} ->
Ash.DataLayer.rollback(Enum.at(resources, 0), error)
end
end,
opts[:timeout] || :infinity,
transaction_metadata(opts)
)
|> case do
{:ok, result} ->
if notify? do
saved_notifications = Process.delete(:ash_notifications) || []
remaining_notifications = result.resource_notifications ++ saved_notifications
{:ok, %{result | resource_notifications: remaining_notifications}}
else
{:ok, result}
end
{:error, error} ->
{:error, error}
end
after
if notify? do
Process.delete(:ash_started_transaction?)
end
end
end
true ->
task_with_timeout(
fn ->
do_run(requests, opts)
end,
opts[:resource],
opts[:timeout],
opts[:name],
opts[:tracer]
)
end
|> case do
{:ok, %{resource_notifications: resource_notifications} = result} ->
notifications =
if Process.get(:ash_started_transaction?) && !opts[:return_notifications?] do
current_notifications = Process.get(:ash_notifications, [])
Process.put(
:ash_notifications,
current_notifications ++ resource_notifications
)
[]
else
resource_notifications
end
{:ok, %{result | resource_notifications: notifications}}
{:error, :timeout} ->
{:error, Ash.Error.Invalid.Timeout.exception(timeout: opts[:timeout], name: opts[:name])}
other ->
other
end
end
@doc false
def task_with_timeout(fun, resource, timeout, name, tracer) do
if !Application.get_env(:ash, :disable_async?) &&
(is_nil(resource) ||
Ash.DataLayer.data_layer_can?(resource, :async_engine)) && timeout &&
timeout != :infinity && !Ash.DataLayer.in_transaction?(resource) do
task =
async(
fun,
tracer: tracer
)
try do
case Task.await(task, timeout) do
{:__exception__, e, stacktrace} ->
reraise e, stacktrace
other ->
other
end
catch
:exit, {:timeout, {Task, :await, [^task, timeout]}} ->
{:error, Ash.Error.Invalid.Timeout.exception(timeout: timeout, name: name)}
end
else
fun.()
end
end
defp transaction_metadata(opts) do
case opts[:transaction_reason] do
%{metadata: metadata} = reason ->
%{reason | metadata: Map.put(metadata, :actor, opts[:actor])}
nil ->
%{
type: :custom,
metadata: %{
actor: opts[:actor]
}
}
other ->
other
end
end
def do_run(requests, opts \\ []) do
state =
%__MODULE__{
ref: make_ref(),
id: Ash.UUID.generate(),
resolved_fields: %{},
actor: opts[:actor],
return_notifications?: opts[:return_notifications?],
failure_mode: opts[:failure_mode] || :stop,
concurrency_limit: System.schedulers_online() * 2,
authorize?: opts[:authorize?] || false,
verbose?: opts[:verbose?] || false,
async?: !Application.get_env(:ash, :disable_async?),
opts: opts
}
|> add_requests(requests)
log(state, fn ->
"Engine Starting - #{Enum.map_join(state.requests, ", ", & &1.name)}"
end)
result = run_to_completion(state)
status =
case result.errors do
[] -> :ok
_ -> :error
end
{status,
result.requests
|> Enum.reduce(result, &add_data(&2, &1.path, &1.data))}
end
defp add_data(state, _path, %Ash.Engine.Request.UnresolvedField{}), do: state
defp add_data(state, path, data) do
%{
state
| data: put_nested_key(state.data, path, data)
}
end
defp set_async(requests) do
if Application.get_env(:ash, :disable_async?) do
Enum.map(requests, &Map.put(&1, :async?, false))
else
Enum.map(requests, fn request ->
if must_be_local?(request) do
%{request | async?: false}
else
request
end
end)
end
end
@doc false
def must_be_local?(%{async?: false}), do: true
def must_be_local?(request) do
[request.resource | request.touches_resources || []]
|> Enum.filter(fn item -> item && is_atom(item) end)
|> Enum.any?(fn resource ->
(Ash.DataLayer.data_layer_can?(resource, :transact) &&
Ash.DataLayer.in_transaction?(resource)) ||
not Ash.DataLayer.data_layer_can?(resource, :async_engine)
end)
end
defp run_to_completion(state) do
if complete?(state) do
log(state, "Engine Complete")
state
else
state
|> run_iteration()
|> case do
^state ->
if state.tasks == [] && state.pending_tasks == [] do
if state.errors == [] do
detect_deadlocks(state)
raise """
Engine Deadlock! No async tasks and state is the same after iteration.
#{long_breakdown(state)}
"""
else
state
end
else
state
|> start_pending_tasks()
|> run_iteration()
|> run_to_completion()
end
new_state ->
run_to_completion(new_state)
end
end
end
defp complete?(%{failure_mode: :continue} = state) do
Enum.all?(state.requests, &(&1.state in [:complete, :error]))
end
defp complete?(%{failure_mode: :stop} = state) do
Enum.all?(state.requests, &(&1.state in [:complete])) ||
Enum.any?(state.requests, &(&1.state in [:error]))
end
defp errors(state) do
if state.errors == [] do
""
else
"""
Errors:
#{Enum.map_join(state.errors, "\n", fn error -> Exception.format(:error, Ash.Error.to_ash_error(error), stacktrace(error)) end)}
"""
end
end
defp stacktrace(%{stacktrace: %{stacktrace: stacktrace}})
when not is_nil(stacktrace) do
stacktrace
end
defp stacktrace(_), do: nil
defp completed_requests(state) do
state.requests
|> Enum.filter(&(&1.state == :complete))
|> case do
[] ->
""
requests ->
"""
Complete Requests:
#{Enum.map_join(requests, "\n\n", fn request -> Request.summarize(request) end)}
"""
end
end
defp errored_requests(state) do
state.requests
|> Enum.filter(&(&1.state == :error))
|> case do
[] ->
""
requests ->
"""
Errored Requests:
#{Enum.map_join(requests, "\n\n", fn request -> Request.summarize(request) end)}
"""
end
end
defp pending_requests(state) do
state.requests
|> Enum.reject(&(&1.state in [:complete, :error]))
|> case do
[] ->
""
requests ->
"""
Requests:
#{Enum.map_join(requests, "\n\n", fn request -> Request.summarize(request) <> "\n " <> depends_on_summary(request, state) end)}
"""
end
end
defp depends_on_summary(%{path: request_path} = request, state) do
dependencies =
state.dependencies_waiting_on_request
|> Kernel.||([])
|> Enum.filter(fn
{^request_path, _} ->
true
_ ->
false
end)
|> Enum.map(fn {_, dep} ->
{:lists.droplast(dep), List.last(dep)}
end)
|> Enum.concat(state.dependencies[request_path] || [])
if Enum.empty?(dependencies) do
" state: #{request.state}"
else
" state: #{request.state} | depends on #{Enum.map_join(dependencies, ", ", &name_of(&1, state))}"
end
end
defp name_of({path, dep}, state) do
case Enum.find(state.requests, &(&1.path == path)) do
nil ->
"unknown dependency: #{inspect(path)} -> #{inspect(dep)}"
request ->
"#{request.name} -> #{inspect(dep)}"
end
end
defp run_iteration(
%__MODULE__{unsent_dependencies: [{request_path, dep} | remaining_unsent_dependencies]} =
state
) do
state = %{state | unsent_dependencies: remaining_unsent_dependencies}
request = Enum.find(state.requests, &(&1.path == request_path))
path = :lists.droplast(dep)
field = List.last(dep)
depended_on_request = Enum.find(state.requests, &(&1.path == path))
cond do
request.state == :error ->
state
depended_on_request.state == :error ->
case Request.wont_receive(request, path, field) do
{:stop, :dependency_failed, new_request} ->
new_request = %{new_request | state: :error}
state
|> replace_request(new_request)
|> notify_error(new_request)
end
true ->
{state, notifications, dependencies} =
notify_local_request(
state,
depended_on_request,
request,
field
)
state
|> notify(notifications)
|> store_dependencies(dependencies)
end
end
defp run_iteration(state) do
state =
state.tasks
|> Enum.reduce(state, fn task, state ->
case Task.yield(task, 0) do
{:ok, {:__exception__, exception, stacktrace}} ->
reraise exception, stacktrace
{:ok, {request_path, result}} ->
state = %{state | tasks: state.tasks -- [task]}
request = Enum.find(state.requests, &(&1.path == request_path))
new_request = %{request | async_fetch_state: {:fetched, result}}
replace_request(state, new_request)
nil ->
state
end
end)
case Enum.find(state.requests, &match?({:requested, _}, &1.async_fetch_state)) do
nil ->
{async, sync} =
state.requests
|> Enum.filter(fn request ->
Enum.empty?(state.dependencies[request.path] || [])
end)
|> Enum.split_with(& &1.async?)
new_state = Enum.reduce(async, state, &do_run_iteration(&2, &1))
if state == new_state do
Enum.reduce(sync, new_state, &do_run_iteration(&2, &1))
else
new_state
end
request ->
{_, resolver_context} = request.async_fetch_state
request = %{request | async_fetch_state: :fetching}
pending_task = fn ->
Ash.Tracer.span :request_step,
request.name,
request.tracer do
metadata = %{
resource_short_name:
request.resource && Ash.Resource.Info.short_name(request.resource),
resource: request.resource,
actor: request.actor,
action: request.action && request.action.name,
authorize?: request.authorize?
}
Ash.Tracer.set_metadata(request.tracer, :request_step, metadata)
Ash.Tracer.telemetry_span [:ash, :request_step], %{name: request.name} do
{request.path, request.data.resolver.(resolver_context)}
end
end
end
replace_request(
%{state | pending_tasks: [pending_task | state.pending_tasks]},
request
)
end
end
@doc false
def async(func, opts) do
ash_context = Ash.get_context_for_transfer(opts)
{:current_stacktrace, stacktrace} = Process.info(self(), :current_stacktrace)
Task.async(fn ->
try do
Ash.transfer_context(ash_context, opts)
func.()
rescue
e ->
e =
if Ash.Error.ash_error?(e) do
if e.stacktrace && e.stacktrace.stacktrace do
update_in(e.stacktrace.stacktrace, &(&1 ++ Enum.drop(stacktrace, 1)))
else
e
end
else
e
end
{:__exception__, e, __STACKTRACE__ ++ Enum.drop(stacktrace, 1)}
end
end)
end
defp do_run_iteration(state, request) do
{state, notifications, dependencies} = fully_advance_request(state, request)
state
|> notify(notifications)
|> store_dependencies(dependencies)
end
defp detect_deadlocks(state) do
state.dependencies
|> Enum.each(fn {path, deps} ->
case Enum.find_value(deps, &depends_on(state, &1, path)) do
nil ->
:ok
{circular_dep, path} ->
raise "Deadlock detected! #{inspect(path)} and #{circular_dep} depend on each other via #{Enum.map_join(path, " -> ", &inspect/1)}"
circular_dep ->
raise "Deadlock detected! #{inspect(path)} and #{circular_dep} depend on each other"
end
end)
state
end
defp depends_on(state, source, destination, trail \\ []) do
deps =
state.dependencies[source]
|> Kernel.||([])
if Enum.any?(deps, fn {path, _} -> path == destination end) do
source
else
deps
|> Enum.reject(&(&1 in trail))
|> Enum.find(&depends_on(state, &1, destination, [&1 | []]))
|> case do
nil ->
false
{circular_dep, path} ->
{source, path ++ circular_dep}
circular_dep ->
{source, [circular_dep]}
end
end
end
def long_breakdown(state) do
"""
#{errors(state)}
#{completed_requests(state)}
#{errored_requests(state)}
#{pending_requests(state)}
"""
end
defp fully_advance_request(state, request) do
case advance_request(request, state) do
{:ok, new_request, notifications, dependencies, resource_notification} ->
new_state =
state
|> replace_request(new_request)
|> add_resource_notification(resource_notification)
new_dependencies = build_dependencies(new_request, dependencies)
{new_state, notifications, new_dependencies}
{:ok, new_request, notifications, dependencies} ->
new_state = replace_request(state, new_request)
new_dependencies = build_dependencies(new_request, dependencies)
{new_state, notifications, new_dependencies}
{:error, error, new_request} ->
new_request = %{new_request | state: :error}
new_state =
state
|> add_error(new_request.path, error)
|> replace_request(new_request)
|> notify_error(new_request)
{new_state, [], []}
end
end
defp notify_error(state, new_request) do
state.reverse_dependencies[new_request.path]
|> Kernel.||([])
|> Enum.reduce(state, fn {request_path_that_wont_receive, dep_they_wont_receive}, state ->
request = Enum.find(state.requests, &(&1.path == request_path_that_wont_receive))
if request.state == :error do
state
else
case Request.wont_receive(
request,
new_request.path,
dep_they_wont_receive
) do
{:stop, :dependency_failed, new_request} ->
new_request = %{new_request | state: :error}
state
|> notify_error(new_request)
|> replace_request(new_request)
end
end
end)
end
defp start_pending_tasks(%{pending_tasks: []} = state), do: state
defp start_pending_tasks(state) do
available_tasks = state.concurrency_limit - Enum.count(state.tasks)
{to_start, remaining} = Enum.split(state.pending_tasks, available_tasks)
state = %{state | pending_tasks: remaining}
new_tasks = Enum.map(to_start, &async(&1, state.opts))
%{state | tasks: state.tasks ++ new_tasks}
end
defp advance_request(%{state: state} = request, _state) when state in [:error] do
{:ok, request, [], []}
end
defp advance_request(request, state) do
case Request.next(request) do
{:already_complete, new_request, new_notifications, new_dependencies} ->
{:ok, new_request, new_notifications, new_dependencies}
{:complete, new_request, new_notifications, new_dependencies} ->
if new_request.resource && new_request.notify? do
resource_notification =
sanitize_notification(Request.resource_notification(new_request), state)
{:ok, new_request, new_notifications, new_dependencies, resource_notification}
else
{:ok, new_request, new_notifications, new_dependencies}
end
{:continue, new_request, new_notifications} ->
{:ok, new_request, new_notifications, []}
{:error, error, new_request} ->
{:error, error, new_request}
{:wait, new_request, new_notifications, new_dependencies} ->
{:ok, new_request, new_notifications, new_dependencies}
end
end
defp notify(state, notifications) do
notifications =
notifications
|> Enum.uniq()
|> Enum.reject(&MapSet.member?(state.notifications, &1))
new_state_notifications = Enum.reduce(notifications, state.notifications, &MapSet.put(&2, &1))
state = %{state | notifications: new_state_notifications}
notifications
|> Request.sort_and_clean_notifications()
|> Enum.reduce(state, fn
{:requests, requests}, state ->
state
|> add_requests(requests)
{:set_extra_data, key, value}, state ->
%{state | data: Map.put(state.data, key, value)}
%Ash.Notifier.Notification{} = resource_notification, state ->
resource_notification = sanitize_notification(resource_notification, state)
add_resource_notification(state, resource_notification)
{receiver_path, request_path, field, value}, state ->
receiver_request = Enum.find(state.requests, &(&1.path == receiver_path))
{:continue, new_request} =
Request.receive_field(receiver_request, request_path, field, value)
state
|> Map.update!(:dependencies, fn dependencies ->
Map.update(dependencies, receiver_request.path, MapSet.new(), fn deps ->
MapSet.delete(deps, {request_path, field})
end)
end)
|> Map.update!(:reverse_dependencies, fn dependencies ->
Map.update(dependencies, request_path, MapSet.new(), fn deps ->
MapSet.delete(deps, {receiver_request.path, field})
end)
end)
|> replace_request(new_request)
end)
end
defp sanitize_notification(resource_notification, state) do
%{
resource_notification
| from: self(),
metadata:
Map.merge(
resource_notification.metadata || %{},
state.opts[:notification_metadata] || %{}
)
}
end
defp store_dependencies(state, dependencies) do
dependencies
|> Enum.uniq()
|> Enum.reject(fn {request_path, dep} = path_and_field ->
MapSet.member?(state.dependencies_seen, path_and_field) ||
dep in (state.dependencies[request_path] || [])
end)
|> Enum.reduce(
state,
fn {request_path, dep} = seen_dep, state ->
dep_path = :lists.droplast(dep)
dep_field = List.last(dep)
depended_on_request = Enum.find(state.requests, &(&1.path == dep_path))
if depended_on_request do
# we want to send things from non async requests
# after we've sent all info to async requests
unsent_dependencies =
if depended_on_request && depended_on_request.async? do
state.unsent_dependencies ++ [{request_path, dep}]
else
[{request_path, dep} | state.unsent_dependencies]
end
%{
state
| dependencies:
state.dependencies
|> Map.put_new_lazy(request_path, fn -> MapSet.new() end)
|> Map.update!(request_path, &MapSet.put(&1, {dep_path, dep_field})),
reverse_dependencies:
state.reverse_dependencies
|> Map.put_new_lazy(dep_path, fn -> MapSet.new() end)
|> Map.update!(dep_path, &MapSet.put(&1, {request_path, dep_field})),
dependencies_seen: MapSet.put(state.dependencies_seen, seen_dep),
unsent_dependencies: unsent_dependencies
}
else
%{
state
| dependencies_waiting_on_request:
MapSet.put(state.dependencies_waiting_on_request, {request_path, dep})
}
end
end
)
end
defp notify_local_request(
state,
depended_on_request,
request,
field
) do
case Request.send_field(depended_on_request, request.path, field) do
{:ok, new_request, new_notifications} ->
state = replace_request(state, new_request)
{state, new_notifications, []}
{:waiting, new_request, new_notifications, new_dependencies} ->
new_dependencies = build_dependencies(new_request, new_dependencies)
new_state = replace_request(state, new_request)
{new_state, new_notifications, new_dependencies}
{:error, error, new_request} ->
new_state =
state
|> replace_request(%{new_request | state: :error})
|> add_error(new_request.path, error)
{new_state, [], []}
end
end
defp add_error(state, path, errors) when is_list(errors) do
Enum.reduce(errors, state, &add_error(&2, path, &1))
end
defp add_error(state, _path, error) do
error = Ash.Error.to_ash_error(error)
if error in state.errors do
state
else
%{state | errors: [error | state.errors]}
end
end
defp replace_request(state, request) do
%{
state
| requests:
Enum.map(state.requests, fn existing_request ->
if existing_request.path == request.path do
request
else
existing_request
end
end)
}
end
defp add_resource_notification(state, resource_notification) do
if Ash.DataLayer.in_transaction?(resource_notification.resource) ||
state.return_notifications? do
%{state | resource_notifications: [resource_notification | state.resource_notifications]}
else
unsent = Ash.Notifier.notify(resource_notification)
%{state | resource_notifications: unsent ++ state.resource_notifications}
end
end
def add_requests(state, requests) do
requests =
Enum.reject(requests, fn new_request ->
Enum.find(state.requests, &(&1.path == new_request.path))
end)
{async, non_async} =
requests
|> Enum.map(fn request ->
if Enum.find(state.requests, &(&1.path == request.path)) do
raise """
Attempted to add request #{inspect(request.path)} but it has already been added!
"""
end
authorize? = request.authorize? and state.authorize?
%{
request
| authorize?: authorize?,
authorized?: !authorize?,
actor: request.actor || state.actor,
verbose?: request.verbose? || state.verbose?,
async?: request.async? and state.async?,
tracer: state.opts[:tracer]
}
end)
|> Enum.map(&Request.add_initial_authorizer_state/1)
|> set_async()
|> Enum.split_with(& &1.async?)
%{state | requests: async ++ state.requests ++ non_async}
|> add_dependencies_waiting_on_request(requests)
end
defp add_dependencies_waiting_on_request(state, new_requests) do
state.dependencies_waiting_on_request
|> Enum.reduce(state, fn
{request_path, dep}, state ->
dep_path = :lists.droplast(dep)
if Enum.any?(new_requests, &(&1.path == dep_path)) do
%{
state
| unsent_dependencies: [{request_path, dep} | state.unsent_dependencies],
dependencies_waiting_on_request:
MapSet.delete(state.dependencies_waiting_on_request, {request_path, dep})
}
else
state
end
end)
end
defp build_dependencies(request, dependencies) do
Enum.map(dependencies, fn dep ->
{request.path, dep}
end)
end
defp log(state, message, level \\ :debug)
defp log(%{verbose?: true, id: id}, message, level) when is_function(message) do
Logger.log(level, fn -> ["#{id}: ", message.()] end)
end
defp log(%{verbose?: true, id: id}, message, level) do
Logger.log(level, fn -> ["#{id}: ", message] end)
end
defp log(_, _, _) do
false
end
def put_nested_key(state, [key], value) do
Map.put(state, key, value)
end
def put_nested_key(state, [key | rest], value) do
case Map.fetch(state, key) do
{:ok, nested_state} when is_map(nested_state) ->
Map.put(state, key, put_nested_key(nested_state, rest, value))
_ ->
Map.put(state, key, put_nested_key(%{}, rest, value))
end
end
def put_nested_key(state, key, value) do
Map.put(state, key, value)
end
end