Hey @shiroyasha,

Checkout:
https://github.com/Nebo15/sage

Here are some arguments to use it:

It’s like Ecto.Multi but across business logic and third-party APIs.

Features

• Transaction retries;
• Asynchronous transactions with timeout;
• Retries with exponential backoff and jitter;
• Ease to write circuit breakers;
• Code that is clean and easy to test;
• Low cost of integration in existing code base and low performance overhead;
• Ability to not lock the database with long running transactions;
• Extensibility - write your own handler for critical errors or metric collector to measure how much time each step took.

Goals of the Sage project:

• Become a de facto tool to run distributed transactions in the Elixir world;
• Stay simple to use and small to maintain: less code - less bugs;
• Educate people how to run distributed transaction pragmatically.

Problematic code:

defmodule WithExample do
  def create_and_subscribe_user(attrs) do
    Repo.transaction(fn ->
      with {:ok, user} <- create_user(attrs),
           {:ok, plans} <- fetch_subscription_plans(attrs),
           {:ok, charge} <- charge_card(user, subscription),
           {:ok, subscription} <- create_subscription(user, plan, attrs),
           {:ok, _delivery} <- schedule_delivery(user, subscription, attrs),
           {:ok, _receipt} <- send_email_receipt(user, subscription, attrs),
           {:ok, user} <- update_user(user, %{subscription: subscription}) do
        acknowledge_job(opts)
      else
        {:error, {:charge_failed, _reason}} ->
          # First problem: charge is not available here
          :ok = refund(charge)
          reject_job(opts)

        {:error, {:create_subscription, _reason}} ->
          # Second problem: growing list of compensations
          :ok = refund(charge)
          :ok = delete_subscription(subscription)
          reject_job(opts)

        # Third problem: how to decide when we should be sending another email or
        # at which stage we've failed?

        other ->
          # Will rollback transaction on all other errors
          :ok = ensure_deleted(fn -> refund(charge) end)
          :ok = ensure_deleted(fn -> delete_subscription(subscription) end)
          :ok = ensure_deleted(fn -> delete_delivery_from_schedule(delivery) end)
          reject_job(opts)

          other
      end
    end)
  end

  defp ensure_deleted(cb) do
    case cb.() do
      :ok -> :ok
      {:error, :not_found} -> :ok
    end
  end
end

Sage’s Solution

defmodule SageExample do
  import Sage
  require Logger

  @spec create_and_subscribe_user(attrs :: map()) :: {:ok, last_effect :: any(), all_effects :: map()} | {:error, reason :: any()}
  def create_and_subscribe_user(attrs) do
    new()
    |> run(:user, &create_user/2)
    |> run(:plans, &fetch_subscription_plans/2, &subscription_plans_circuit_breaker/3)
    |> run(:subscription, &create_subscription/2, &delete_subscription/3)
    |> run_async(:delivery, &schedule_delivery/2, &delete_delivery_from_schedule/3)
    |> run_async(:receipt, &send_email_receipt/2, &send_excuse_for_email_receipt/3)
    |> run(:update_user, &set_plan_for_a_user/2)
    |> finally(&acknowledge_job/2)
    |> transaction(SageExample.Repo, attrs)
  end

  # Transaction behaviour:
  # @callback transaction(attrs :: map()) :: {:ok, last_effect :: any(), all_effects :: map()} | {:error, reason :: any()}

  # Compensation behaviour:
  # @callback compensation(
  #             effect_to_compensate :: any(),
  #             effects_so_far :: map(),
  #             attrs :: any()
  #           ) :: :ok | :abort | {:retry, retry_opts :: Sage.retry_opts()} | {:continue, any()}

  def create_user(_effects_so_far, %{"user" => user_attrs}) do
    %SageExample.User{}
    |> SageExample.User.changeset(user_attrs)
    |> SageExample.Repo.insert()
  end

  def fetch_subscription_plans(_effects_so_far, _attrs) do
    {:ok, _plans} = SageExample.Billing.APIClient.list_plans()
  end

  # If we failed to fetch plans, let's continue with cached ones
  def subscription_plans_circuit_breaker(_effect_to_compensate, _effects_so_far, _attrs) do
    {:continue, [%{"id" => "free", "total" => 0}, %{"id" => "standard", "total" => 4.99}]}
  end

  def create_subscription(%{user: user}, %{"subscription" => subscription}) do
    {:ok, subscription} = SageExample.Billing.APIClient.subscribe_user(user, subscription["plan"])
  end

  def delete_subscription(_effect_to_compensate, %{user: user}, _attrs) do
    :ok = SageExample.Billing.APIClient.delete_all_subscriptions_for_user(user)
    # We want to apply forward compensation from :subscription stage for 5 times
    {:retry, retry_limit: 5, base_backoff: 10, max_backoff: 30_000, enable_jitter: true}
  end

  # .. other transaction and compensation callbacks

  def acknowledge_job(:ok, attrs) do
    Logger.info("Successfully created user #{attrs["user"]["email"]}")
  end

  def acknowledge_job(_error, attrs) do
    Logger.warn("Failed to create user #{attrs["user"]["email"]}")
  end
end

Result

Along with a readable code, you are getting:

• Reasonable guarantees that all transaction steps are completed or all failed steps are compensated;
• Code which is much simpler and easier to test a code;
• Retries, circuit breaking and asynchronous requests out of the box;
• Declarative way to define your transactions and run them.

P.S. I am not super experienced with Elixir, but this library seemed useful so I kept it in my bookmarks.

Loading things concurrent isn’t always as good as it sounds. It’s possible it makes things lot worse. Just calling them sequentially and let different request basically do concurrent request to the database might be better because it’s more fair sharing of database resource between requests. We did this in the past and had to refactor our app away from it because one request would start lot of multiple data calls to database and block short request with only few database call that arrived little bit later. So I would think hard before doing this kind of “optimization”.

@Eiji Thanks once again for the great breakdown. The organize_nested is definitely hitting on many things I’m looking for. I’ll try to extract the essence and generalize the snippet that you shared.

@wanton7 good reminder, and thank you for sharing your path of doing this and then refactoring it to a fully sequential approach. I hit this same concern in ~ 2020 when we introduced a lot more parallelism into our codebase to reduce the processing time. I have strong signals from the last three years (metrics & traces) that show that parallelism significantly improves overall performance.

How can this be? We are still hitting the same database, right?

It seems that two factors make this assumption incorrect.

First, most resources are heavily cached, with level 1 being the local in-memory cache, level 2 being cluster distributed cache or dedicated cache storage like Redis, and level 3 being the database. So a typical request would hit multiple independent data systems.

Secondly, not all of our requests are database bound. We are a CI/CD provider where internal and external resources like virtual machines, docker registries, blob storage, and job processing units provide their internal APIs.

Sequential vs. Fully Concurrent
These are the two extremes. Loading things one-by-one or spawning up concurrent tasks without limits. In my experience, both approaches are suboptimal. The first one doesn’t take advantage of available resources when they are readily available, while the second can introduce more harm than good, like in the example you shared.

Ideally, I would like to express the dependencies and let the system decide when it is best to go in parallel and when it is best to go one-by-one. Something like (pseudocode):

resources = [
  {:a, &load_a/1},
  {:b, &load_b/1},
  {:c, &load_c/2, deps: [:a, :b]}
]

load(resources) # the load function now knows the dependencies and should be able to make the best decision based on the system's state and the platform's saturation.

Once again, thank you all for sharing your input. It helps me a lot to clarify my thoughts and question my assumptions. I’ll make sure to share which direction my team will take and, of course, share some open-source code and examples.