Dealing with concurrent updates in database

thiagopmac

I’m building a Phoenix server that handles financial transactions, It’s very simple it creates an account, creates a user, and processes transactions between users. The question that I have is related to concurrency, to give more context the code below creates a transaction:

  def create_transaction(%{sender_id: sender_id, recipient_id: recipient_id, amount: amount}) do
    sender_update_query =
      from Account,
        where: [id: ^sender_id],
        update: [inc: [balance: ^(-amount)]]

    recipient_update_query =
      from Account,
        where: [id: ^recipient_id],
        update: [inc: [balance: ^(+amount)]]

    Multi.new()
    |> Multi.run(:retrieved_accounts, &retrieved_accounts(&1, &2, recipient_id, sender_id))
    |> Multi.run(:check_sender_funds, &check_sender_funds(&1, &2, amount))
    |> Multi.update_all(:recipient_update_query, recipient_update_query, [])
    |> Multi.run(:check_recipient_update_query, &check_recipient_update_query(&1, &2))
    |> Multi.update_all(:sender_update_query, sender_update_query, [])
    |> Multi.run(:check_sender_update_query, &check_sender_update_query(&1, &2))
    |> Multi.insert(:insert_transaction, %Transaction{
      sender_id: sender_id,
      recipient_id: recipient_id,
      amount: amount
    })
    |> Repo.transaction()
    |> handle_multi()
  end

  defp retrieved_accounts(_repo, _changes, recipient_id, sender_id) do
    id_list = [sender_id, recipient_id]
    Accounts.get_sender_and_recipient_accounts(id_list, sender_id, recipient_id)
  end

  defp check_sender_funds(_repo, %{retrieved_accounts: [sender_account, _]}, amount) do
    if sender_account.balance - amount >= 0,
      do: {:ok, nil},
      else: {:error, :insufficient_funds}
  end

  defp check_recipient_update_query(
         _repo,
         %{recipient_update_query: {1, _}}
       ) do
    {:ok, nil}
  end

  defp check_recipient_update_query(
         _repo,
         %{recipient_update_query: {_, _}}
       ) do
    {:error, :failed_transfer}
  end

  defp check_sender_update_query(
         _repo,
         %{sender_update_query: {1, _}}
       ) do
    {:ok, nil}
  end

  defp check_sender_update_query(_repo, %{sender_update_query: {_, _}}) do
    {:error, :failed_transfer}
  end

  defp handle_multi({:ok, %{insert_transaction: transaction}}), do: {:ok, transaction}
  defp handle_multi({:ok, %{update_transaction: transaction}}), do: {:ok, transaction}
  defp handle_multi({:error, _id, error_or_changeset, _multi}), do: {:error, error_or_changeset}

My question is if this function will handle cases where multiple transactions can happen at the same time.

Another thing that I should mention is that I also have a function that charges back the transaction, which means that the balance can be modified by another function. I don’t know if Ecto can handle this kind of operation or if I should try to use something like a Genserver to orchestrate the transactions. Thoughts?

2 comments

/phoenix #ecto #concurrency #genserver

0 1051 2

2023-08-23 14:26:48 UTC

First 2 of 2 Posts

al2o3cr

This is mostly a general SQL question - Ecto provides some helpers, but the patterns are universal across a lot of ORMs.

There are two main strategies:

“pessimistic” locking: use the database’s locking mechanisms (SELECT ... FOR UPDATE etc) to protect the rows before they are modified. Your code will need to deal with problems like deadlock - for instance, what should happen if Alice and Bob try to send money to each other simultaneously. Check out Ecto.Query.lock and your DB"s docs (link is for PG) for more detail.
“optimistic” locking: instead of using locks to ensure that nobody else writes to the rows, this adds a “version” to the rows and checks it on update. If something else has updated the row, the version won’t match and the update will need to be retried. See Ecto.Changeset.optimistic_lock for more detail.

Post #1

hubertlepicki

generally speaking, RDBMS systems have two mechanisms: transactions and locks. They can be used separately, but they can also be used together.

Transaction is a mechanism that ensures that all statements in the database will execute, and all records will be created/deleted/updated. If there’s an error, the whole transaction is rolled back.

Locks is another mechanism, where you can limit, or prevent altogether, many concurrent queries to happen at the same time.

In your example, you have transaction. It’s still possible that another transaction executes concurrently, and, for example: both transactions succeed because their :check_sender_funds steps will succeed, but at the end of the second transaction you’ll have some unexpected money missing somewhere.

In order to prevent that, you can use locking. If you use PostgreSQL with Ecto, you can customize my code, that wraps code in transaction and lock. You could probably make it work with Ecto.Multi, by making the lock query set up lock as first statement in the transaction:

Repo.transaction(fn ->
  Repo.query("SELECT pg_advisory_xact_lock(1)")

  # put the rest of the code that makes db changes or queries db here:
  ....
end)

You can have multiple locks at the same time across the system, I am using “1” as the lock key just for this example, but this can be user ID, tenant ID or some other resource ID (like account ID) that you want to have the lock unique for.

This works in the way that if another transaction opens up that wants to set up the lock on the same resource, it will wait on the lock line until the other transaction finishes.

You can also use the locking mechanisms described above by @al2o3cr depending on what you think is best.

Post #2

— All posts loaded —