[draft] Loader

[shiroyasha]

One pattern I encounter all around our code base is loading resources in parallel and preparing the response based on the loaded resources.

Here is a toy example (not a real code, just an example):

with(
  load_org_task <- Task.async(fn -> load_org(org_id) end),
  load_user_task <- Task.async(fn -> load_user(user_id) end),
  load_project_task <- Task.async(fn -> load_project(user_id) end),
  {:ok, org} <- Task.await(load_org_task),
  {:ok, user} <- Task.await(load_user_task),
  {:ok, project} <- Task.await(load_project_task),
  load_artifact_list_task <- Task.async(fn -> list_arifacts(org_id) end),
  load_permissions_task <- Task.async(fn -> load_permissions(project) end),
  {:ok, artifact_list} <- Task.await(load_artifact_list_task),
  {:ok, permissions} <- Task.await(load_permissions)
) do
  render(org, user, project, artifact_list, permissions)
else
  {:error, :not_found} -> ...
end

Things I don't like about this:

• Error handling is a mess; we don't have a clear and simple way to figure out what failed.
• Defining dependencies between loaded resources is manual and sometimes not obvious
• Timeouts, fail-fast, and observability are ad-hoc created

Loader — Streamlining the loading process

In this PR (the code is not yet ready, just the idea), I'm introducing a utility to help with these tasks. For lack of a better word, I'm calling it a "loader".

Here is how it would work for the above example:

results = Loader.load([
  {:org, &load_org/2, [org_id]},
  {:user, &load_user/2, [user_id]},
  {:project, &load_project/2, [project_id]},
  {:artifacts, &load_artifacts/2, depends_on: [:project]},
  {:permissions, &load_permissions/2, depends_on: [:user, :project, :org]}
])

case results do
  {:ok, resources} -> render(resources)
  
  {:error, {:user, {:error, :not_found}} -> render("user not found")
  {:error, {:org, {:error, :not_found}} -> render("org not found")
  e -> 
end

What is included?

Arbitrary resource dependency trees loaded in parallel

The loader will load things in parallel and wait for dips where it must.
Example:

Loader.load([
  {:a, &load_a/2}
  {:b, &load_b/2, depends_on: [:a]},
  {:c, &load_c/2, depends_on: [:a]},
  {:d, &load_d/2, depends_on: [:b, :c]}
])

Timeouts
You can set up timeouts for individual resources and the whole load operation.
Example:

opts = %{
  whole_operation_timeout: 30_000,
  per_resource_timeout: 5_000
}

Loader.load([
  {:a, &load_a, timeout: 1000}
  {:b, &load_b}
], opts)

Observability

opts = %{trace: true}

result = Loader.load([
  {:a, &load_a/2, timeout: 1000}
  {:b, &load_b/2}
], opts)

case result do
 {:ok, resources, trace} ->
   Watchman.submit("loading_duration", :timing, trace.total_duration)
 {:error, _, trace} ->
end