mjadczak

Using Ecto to run a long-running multi-process transaction

I’m building an app which provides a read-only GraphQL API to a React frontend. The app stores details of many after-school programs in a certain area.

The actual program data is stored and managed in an external system (Airtable). Once in a while, an import is run to replace all of the data in the Elixir app with the fresh data. Because the data needs to be processed, geolocated etc, and there are potentially tens of thousands of rows, I don’t want to process all of the data as a huge chunk in memory. I will be using either Task.async_stream or GenStage to do the processing.

However, I want to run the whole process (drop all current data, insert new data) in a Postgres transaction. I know truncating the tables will place a read lock on them too, and indeed access to the API will be disabled altogether while the import is running. However I want the option to rollback if errors are encountered during the process.

The only tools Ecto provides are the two variants of Repo.transaction/2. I cannot use Multi since the whole point is to not store all of the data in memory at once. However while playing around with the function variant, I discovered that any Repo calls must be made within that function (or more specifically within that process while the function is being executed), since it is the process that locks the connection and sets up the transaction. Therefore processing the data asynchronously with different processes calling DB functions does not work as expected—any Repo functions called in a different process will not be part of the transaction.

The immediate “shape” of the solution to this kind of serialisation problem is usually a GenServer, but there is no way to separately start and stop the transaction, nor is there a way (without reimplementing/forking :gen_server) to run the GenServer loop inside a transaction function. I came up with this solution instead:

defmodule PFServer.TransactionManager do
  alias PFServer.Repo
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  def execute(fun) when is_function(fun) do
    GenServer.call(__MODULE__, {:execute, fun})
  end

  def finish do
    GenServer.stop(__MODULE__)
  end


  # callbacks

  def init([]) do
    {:ok, pid} = start_link_executor()

    {:ok, pid}
  end

  def handle_call({:execute, fun}, from, pid) do
    send pid, {:execute, fun, from}

    {:noreply, pid}
  end

  def terminate(_reason, pid) do
    send pid, :finish
  end

  # Task execution process
  defp start_link_executor do
    Task.start_link(&execute_until_done/0)
  end

  defp execute_until_done do
    Repo.transaction(&execute_all/0)
  end

  defp execute_all do
    receive do
      {:execute, fun, from} ->
        GenServer.reply from, fun.()
        execute_all()

      :finish -> :ok
    end
  end
end

and now I can spin this up and use TransactionManager.execute(<repo accessing fn>).

Implementing my own receive loop seems like a code smell to me, maybe it’s the learned reliance on OTP to handle any and all messaging concurrency.

Is this an acceptable solution? Am I going to run into problems with this? Is there any other way to handle this? It seems like the OTP way would be to just have the GenServer handle the whole serialization, but like I mentioned this would require manual control over transaction begin and commit.

28 comments

#ecto #databases #genserver

2 8848 28

2017-04-13 00:20:29 UTC

First 10 of 28 Posts

outlog

I would version all the data rows, eg. import_version 1 on all rows..

then you can import/process/insert import_version 2 in the most efficient way and then bump your graphql resolver to select where import_version 2 when everything has been verified to be ok..

you can manage the import_version integer in (d)ets or in it’s own db table (wrap it with concache or something that will cache it in ets/memory) - this way you can also rollback, or go back in time and identify errors etc.

as a rule I would never delete anything in a DB.

Post #1

benwilson512

Author of Craft GraphQL APIs in Elixir with Absinthe

Yeah I like the versioned approach a lot here. It lets you easily import all the new data and then seamlessly switch to using it, you could even easily revert if you had to for some reason.

If space is a concern you can eventually purge old versions as necessary.

Post #2

mjadczak

Since we have a canonical copy of the data in a separate store, I’m not too concerned with retaining previous versions of the data. Importing the data into a second copy of the tables and then switching over would indeed work, but would require much more logic, as well as manual cleanup (if there is an error, have to truncate the half-imported tables manually instead of just rolling back the transaction).

In any case, the issue of how to do a large transaction in Ecto which may involve more than one process remains, and there may be other use cases for this. The code I have now works, and save for perhaps adding some timeout handling for robustness, seems like it does the job—I was just wondering if there was a more straightforward solution.

Post #3

bcahya

Hi mjadczak,

Based on your idea, we sketch our own dirty handling transaction code. Since, we need to ensure that these transaction handling should be:

Compliance with try, rescue, after; cause we should reset our own state through ‘after’. With your genserver approach, our ‘after’ code didn’t executed while raise some exception, leaving our state inconsistent. I didn’t know why
Can easily acquire another transaction

Here our dirty trial code

defmodule Fintech.Transaction do
    alias Fintech.Repo

    def open do
        Repo.transaction &loop/0
    end

    def execute(pid, fun) when is_pid(pid)
        and is_function(fun) do
        send pid, {:execute, fun}
    end

    def rollback(pid) when is_pid(pid) do
        send pid, :rollback    
    end

    def close(pid) when is_pid(pid) do
        send pid, :close
    end

    defp loop do
        receive do
            {:execute, fun} ->
                apply fun, []
                loop()
            :rollback ->
                Repo.rollback :error
                :ok
            :close -> :ok
            _ ->
                Repo.rollback :error
                :ok
        end
    end
end

Sample of executed

  def test_trans do
    pid = spawn Transaction, :open, []
    // set our own state
    try do
      Transaction.execute pid, our_db_trans
      Transaction.execute pid, other_db_trans
      Transaction.execute pid, other_db_trans
      Transaction.close pid
    after
      IO.puts "after executed"
      // set our own state
    rescue
      RuntimeError ->
        Transaction.rollback pid
        IO.puts "rescue runtime error executed"
      _ ->
        Transaction.rollback pid
        IO.puts "other rescue executed"
    end
  end

Post #4

aseigo

Since you are already looking at GenStage .. here’s how I might write this using Flow:

defmodule ExtoTest do
  use Ecto.Repo, otp_app: :ectotest

  def flowing_inner_trans(input) do
    IO.puts "INNER: Are we in a transaction here? #{ExtoTest.in_transaction?}"
    input
  end

  def proc_query(query) do
    IO.puts "OUTER: re we in a transaction here? #{ExtoTest.in_transaction?} #{query}"
  end

  def flowing do
    1..30
    |> Flow.from_enumerable(max_demand: 4)
    |> Flow.each(&ExtoTest.flowing_inner_trans/1)
    |> Stream.each(&ExtoTest.proc_query/1)
    |> Stream.run
  end                             
                                  
  def multi_proc_transactions do  
    ExtoTest.transaction(&ExtoTest.flowing/0)
  end
end

This handles the syncronization for “free” thanks to the Stream.each at the end without ruining the laziness of it. Note that for the above, since it is just producing 30 values, I set the max_demand to something artificially low so you can actually see the async processing:

iex(17)> ExtoTest.multi_proc_transactions

12:30:41.696 [debug] QUERY OK db=0.7ms queue=0.1ms                                                                                                                                                                                           
begin []
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 1
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 2
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 5
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 6
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 9
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 10
OUTER: re we in a transaction here? true 13
OUTER: re we in a transaction here? true 14
OUTER: re we in a transaction here? true 3
OUTER: re we in a transaction here? true 4
OUTER: re we in a transaction here? true 7
OUTER: re we in a transaction here? true 8
OUTER: re we in a transaction here? true 11
OUTER: re we in a transaction here? true 12
OUTER: re we in a transaction here? true 15
OUTER: re we in a transaction here? true 16
OUTER: re we in a transaction here? true 17
OUTER: re we in a transaction here? true 18
OUTER: re we in a transaction here? true 19
OUTER: re we in a transaction here? true 20
OUTER: re we in a transaction here? true 21
OUTER: re we in a transaction here? true 22
OUTER: re we in a transaction here? true 23
OUTER: re we in a transaction here? true 24
OUTER: re we in a transaction here? true 25
OUTER: re we in a transaction here? true 26
OUTER: re we in a transaction here? true 27
OUTER: re we in a transaction here? true 28
OUTER: re we in a transaction here? true 29
OUTER: re we in a transaction here? true 30

12:30:41.708 [debug] QUERY OK db=0.6ms
commit []
{:ok, :ok}

To me, this is rather nicer than setting up a process manually.

iex(17)> ExtoTest.multi_proc_transactions

12:30:41.696 [debug] QUERY OK db=0.7ms queue=0.1ms
begin
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 1
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 2
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 5
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 6
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 9
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
INNER: Are we in a transaction here? false
OUTER: re we in a transaction here? true 10
OUTER: re we in a transaction here? true 13
OUTER: re we in a transaction here? true 14
OUTER: re we in a transaction here? true 3
OUTER: re we in a transaction here? true 4
OUTER: re we in a transaction here? true 7
OUTER: re we in a transaction here? true 8
OUTER: re we in a transaction here? true 11
OUTER: re we in a transaction here? true 12
OUTER: re we in a transaction here? true 15
OUTER: re we in a transaction here? true 16
OUTER: re we in a transaction here? true 17
OUTER: re we in a transaction here? true 18
OUTER: re we in a transaction here? true 19
OUTER: re we in a transaction here? true 20
OUTER: re we in a transaction here? true 21
OUTER: re we in a transaction here? true 22
OUTER: re we in a transaction here? true 23
OUTER: re we in a transaction here? true 24
OUTER: re we in a transaction here? true 25
OUTER: re we in a transaction here? true 26
OUTER: re we in a transaction here? true 27
OUTER: re we in a transaction here? true 28
OUTER: re we in a transaction here? true 29
OUTER: re we in a transaction here? true 30

12:30:41.708 [debug] QUERY OK db=0.6ms
commit
{:ok, :ok}

That said .. if you do stick with your example as above, I would not use Task but create a GenServer module with a handle_info in there. It is, at least imho, cleaner / easier to read than scattering receive blocks in functions that are then passed around via “arbitrary” process spawnings (such as Task)

Post #5

bcahya

Is there any chance to make it commited, lets say on each 10 records ?

Post #6

mjadczak

@bcahya You’re right, my code was in no way a general-purpose solution to the problem I posed, and indeed I have since modified it to add a little more error handling, and I imagine some more will be needed after testing. It’s a pretty narrow-purpose piece of code, but I did want to start a discussion around the larger problem.

@aseigo nice, I didn’t think of using streams that way. Indeed your solution means we can compute the operations in other processes and then serialise them in the current process running the transaction with Stream.each. The problem I have there is that my current solution doesn’t map clearly to that model—I need to do database access during the pre-processing of the data, because there are certain fields (call them categories) which are denormalised in the source data but which I am storing normalised (so I need to check if the category already exists, and if so, fetch its ID for the association).

Regarding your point about Tasks/GenServer—if it were possible to achieve this with GenServer, I would only need one GenServer in the first place. This was exactly what I was complaining about. In this case you have to write your own receive loop somehow, because the receive loop itself must run inside the transaction lambda. With GenServer, when we call GenServer.start_link, a new process is spawned, running the standard GenServer receive loop, and user code is only called on receipt of a message. We would need a way to “wrap” the GenServer loop in the transaction for this to work correctly.

As I pointed out, this could be solved cleanly by allowing decoupled access to transaction state—so that in init/1 we can put the connection of the current process into a transaction state, and then later on, either while handling messages or when terminating, manually commit or rollback the transaction, and perhaps start a new one.

I did take a look at the Ecto code which handles this, and it seems that the pattern of transaction operations being lambdas is baked in all the way down to the adapter code (Multi actually gets run within a lambda through the same process) so I’m not sure how feasible it would be to implement this.

Perhaps @josevalim or another core member could comment? Was there just a decision made to make the interface to transactions a lambda, or is there a fundamental reason why we cannot decouple the begin and commit/rollback? (Read: is it a case of writing a PR which provides a different interface to the underlying connection state management, or would it require a fundamental shift in some assumptions?)

Post #7

aseigo

Try with Stream.scan instead of Stream.each; that should give you a counter (using a number for the accumulator) that can be reset every N iterations, and which would trigger the creation of a new transaction and the closing the existing one (if any)

Post #8

aseigo

If this requires access to data being set within the transaction, then indeed the code can not just wait until the final query to commit. If those reads can be done outside the transaction, then it’s fine as-is; however I assume that if one entry causes the creation of new categories that subsequent entries should re-use those created categories.

That implies another point of serialization through the transaction. Perhaps sth like:

IOW: break it into a multi-stage process where the datafields which should be normalized are first identified and that gets passed into a serialized Stream which does the DB interaction within the transaction. You would probably be passing tuples at this point such as { parsed_data, normalizations_needed } to keep your parsing sensible.

Caching the results of queries in the “fulfil_normalizations” stage could probably also be added to prevent repeated queries for the same data by using sth like Stream.transform and a map as the accumulator to store the cached returns.

It would work exactly as your code stands now, except instead of creating a Task and a separate receive loop, the code would instead start a GenServer and set the pid contained in PFServer.TransactionManager’s state to that. That would allow you to replace the hand-coded receive with handle_* calls in the GenServer, with the :finish call result in a return of {:stop, :normal, .. } to stop the GenServer (and close the transaction on its way out). You still need one GenServer process per transaction you have going on, but at least you don’t have to write a receive loop by hand (with all the possible edge conditions that may be involved).

So it doesn’t solve your main problem, but it is a slightly nicer way to write your example.

Post #9

mjadczak

Doing the synchronisation in multiple stages like this would require at least two passes on each entry of the data, which I’d like to avoid, although it does seem like a neat solution worth investigating.

This won’t work. GenServer.start and GenServer.start_link spawn a separate process in which the GenServer code is run, then return. Doing

Repo.transaction fn -> GenServer.start_link(...) end

will start a transaction, start the GenServer, then immediately commit an empty transaction. All Repo accesses which are supposed to be part of the transaction must synchronously run within that lambda.

On the other hand, we cannot run Repo.transaction anywhere else in the GenServer callbacks, since we wouldn’t actually return back to handle more incoming messages until the transaction lambda returns.

What we need is to wrap the GenServer’s internal receive loop itself with the transaction. This isn’t possible with the Elixir API, but having dived into the Erlang docs, it seems that spawning a new process which, inside a transaction lambda, executes :gen_server.enter_loop may actually do exactly that. We do lose some of the advantages of using start(_link), and we have to register and link the process ourselves, but it may be the best solution—as you mentioned, trying to reïmplement the various edge-case and error-case handling within GenServer is dangerous at best.

I guess the whole thing is a side-effect of the decision in Ecto to hide the concept of connection state from the user, and instead pin connections to processes. At the same time, especially with things like transactions, connection state is important, and making it magic leads to things like this. Perhaps a way in Ecto to explicitly obtain and use a particular connection from the pool would be advisable?

I’ll likely implement the GenServer wrapped-loop solution, think on this some more and perhaps start an issue in Ecto to discuss the underlying architecture more.

Post #10

Load more posts

Where Next?

View thread on forum (has 28 replies!)

ecto

databases

genserver

Home Questions & Help>Questions

#ecto #databases #genserver

21 8843 28

Last post

REPLY VIEW THREAD WITH 28 POSTS

Questions & Help>Questions

How to make first contribution to Phoenix project?

Questions & Help>Questions

Elixir-google-api deprecated, reason and alternatives?

Questions & Help>Questions

One Repo and one database with per-app migrations in an Umbrella project

Questions & Help>Questions

How to integrate Hologram with existing LiveView app?

Questions & Help>Questions

How to do proper post-release steps

Questions & Help>Questions

How to start Oban, but delay job processing?

Questions & Help>Questions

Simple/light way to self host microapps and services?

Questions & Help>Questions

Help with load testing a custom protocol over TCP

Questions & Help>Questions

Why does <.form> discard method and CSRF token when :action is absent, and is action="" a reliable way to submit to the current URL?

Questions & Help>Questions

Help with elixir-ts-mode in doom-emacs config

Questions & Help>Questions

Questions Questions ❯

Latest on Elixir Forum

Hologram v0.11: Regular Expressions, Client Stacktraces, and More

News>News & Updates

Voyager - a modern alternative to Observer

News>Announcing

Beyond Safe Migrations: Better Practices for Ecto + PostgreSQL - Joe Harrow | ElixirConf US

Learning Resources>Talks

ActiveMemory v0.8 — in-memory tables that speak Ecto (schemas, changesets, Repo-style reads)

News>News & Updates

PaperForge - a pure Elixir platform for building, securing, and transforming PDFs

News>Announcing

How to make first contribution to Phoenix project?

Questions & Help>Questions

Talk prep ask: how do you use (or think about) AI agent harnesses? Also migrating an orchestrator from bash to Elixir

Chat & Discussions>AI / LLMs

Patch Package OTP 29.0.5 Released

News>Erlang News

Patch Package OTP 28.5.0.5 Released

News>Erlang News

Patch Package OTP 27.3.4.16 Released

News>Erlang News

Latch - an atproto OAuth client library

News>Announcing

2026/08/06 - Elixir Meetup - Melbourne, Australia

Events/Confs/Meet Ups>List

Elixir's Configuration Layers: A Visual Journey - Stephanie Lane | ElixirConf US

Learning Resources>Talks

Help SWAR optimize URI.encode_www_form/1

Questions & Help>Thoughts On...

Elixir-google-api deprecated, reason and alternatives?

Questions & Help>Questions

Elixir Forum ❯

Sub Categories:

Forums

We're in Beta

About us Mission Statement

Options

Show Best Posts Skip Thread Previews

Using Ecto to run a long-running multi-process transaction

mjadczak

Using Ecto to run a long-running multi-process transaction

First 10 of 28 Posts

outlog

benwilson512

mjadczak

bcahya

aseigo

bcahya

mjadczak

aseigo

aseigo

mjadczak

Where Next?

Trending in Questions

Durable Workflow Execution (temporal alternative for elixir)

Looking for a Prawn-Like PDF Generation Library in Elixir

Simple/light way to self host microapps and services?

Using Phoenix.LiveView.TagEngine as an EEx.Engine is deprecated!

How to integrate Hologram with existing LiveView app?

How to do proper post-release steps

How to start Oban, but delay job processing?

Other Trending Topics

Dexter - A fast, full-featured Elixir LSP optimized for large codebases

Beam Bots - Resilient Robotics on the BEAM

Emerge & Solve - a GUI framework for Elixir

Emily - A new MLX-based backend for Nx

Elixir-lang.org redesign

Programming Nerves (self-published)

Questions & Help>Questions

Latest on Elixir Forum

Categories:

Sub Categories:

Forums

Popular Tags

We're in Beta

Options

Using Ecto to run a long-running multi-process transaction

mjadczak

Using Ecto to run a long-running multi-process transaction

First 10 of 28 Posts

outlog

benwilson512

mjadczak

bcahya

aseigo

bcahya

mjadczak

aseigo

aseigo

mjadczak

Where Next?

Trending in Questions

Durable Workflow Execution (temporal alternative for elixir)

Looking for a Prawn-Like PDF Generation Library in Elixir

Simple/light way to self host microapps and services?

Using Phoenix.LiveView.TagEngine as an EEx.Engine is deprecated!

How to integrate Hologram with existing LiveView app?

How to do proper post-release steps

How to start Oban, but delay job processing?

Other Trending Topics

Dexter - A fast, full-featured Elixir LSP optimized for large codebases

Beam Bots - Resilient Robotics on the BEAM

Emerge & Solve - a GUI framework for Elixir

Emily - A new MLX-based backend for Nx

Elixir-lang.org redesign

Programming Nerves (self-published)

Questions & Help>Questions

Latest on Elixir Forum

Sponsor Spotlight

Our Sponsors

Categories:

Sub Categories:

Forums

Popular Tags

Our Sponsors

We're in Beta

Options