problem asynchronizing ecto calls

Are you running this code in tests with Ecto.Adapters.SQL.Sandbox?

I’m running from mix test with

config :project, Project.Repo,
  pool: Ecto.Adapters.SQL.Sandbox,

in my config/test.exs so I take it I am.

You can either use :shared mode with Ecto.Adapters.SQL.Sandbox and disable async tests for your test module or explicitly transfer the Ecto connection to the spawned task process. See Ecto.Adapters.SQL.Sandbox — Ecto SQL v3.14.0 for more details.

It looks like the project is already doing that. The test file starts with

defmodule Project.Module do
    use Project.ModelCase

And in model_case.ex we’ve got

  setup tags do
    :ok = Ecto.Adapters.SQL.Sandbox.checkout(Project.Repo)

    unless tags[:async] do
      Ecto.Adapters.SQL.Sandbox.mode(Project.Repo, {:shared, self()})
    end

    :ok
  end

So since :async isn’t included in the use line I believe that means we’re using shared mode, so any process should be able to connect to the DB connection.

I’m not sure how relevant it is, but while searching for Adapters.SQL.Sandbox in the project I discovered that it shows up in test/test_helpers.ex:

ExUnit.configure formatters: [JUnitFormatter, ExUnit.CLIFormatter]

ExUnit.start(exclude: [
  :skip,
])

Faker.start()

Ecto.Adapters.SQL.Sandbox.mode(Project.Repo, :manual)

Which seems to set the mode without setting :shared

Try spawning your tasks with Task.async_stream. If that works, then I think the problem is that in your original code task processes are started by a supervisor process and that’s why they don’t see the DB connection checked out by the test process. Performing a manual allowance in the task process should work.

There’s another option now: update your deps and it should just work - Plataformatec on X: "We have just released DBConnection v2.0.4 (which powers Ecto) to use Elixir's v1.8 task callers feature. This means using the SQL Sandbox to test code that uses tasks for concurrency should work out of the box - and it should work concurrently too (wow, such concurrency)!" / X

Thanks for the suggestions. I just tried with async_stream:

    parent = self()
    stream = Task.Supervisor.async_stream(
      Module.TaskSupervisor,
      enum,
      fn (item) -> my_func(item, 2) end,
      [ordered: false, max_concurrency: 1]
    )
    results = Enum.to_list(stream)

then with a manual allow:

    parent = self()
    stream = Task.Supervisor.async_stream_nolink(
      Module.TaskSupervisor,
      enum,
      fn (item) ->
        Ecto.Adapters.SQL.Sandbox.allow(Repo, parent, self())
        my_func(item, 2) end,
      [ordered: false, max_concurrency: 1]
    )
    results = Enum.to_list(stream)

as well as with both:

    parent = self()
    stream = Task.Supervisor.async_stream(
      Module.TaskSupervisor,
      enum,
      fn (item) ->
        Ecto.Adapters.SQL.Sandbox.allow(Repo, parent, self())
        my_func(item, 2) end,
      [ordered: false, max_concurrency: 1]
    )
    results = Enum.to_list(stream)

all unfortunately to no noticeable effect:
14:12:25.494 [error] Postgrex.Protocol (#PID<0.658.0>) disconnected: ** (DBConnection.ConnectionError) owner #PID<0.1745.0> exited while client #PID<0.1748.0> is still running with: shutdown

As for the dependency update route, that looks like a great idea. Alas we’re still on Elixir 1.6 in this project, and I think it would be a fairly large undertaking to get everything working under 1.8. I think we should probably do it at some point, but I don’t think now is the time.

Oops, it looks like I got your suggestion wrong, and had switched from Task.Supervisor.async_stream_nolink to Task.Supervisor.async_stream instead of Task.async_stream.

I’ve now tried switching to Task.async_stream both with a manual allow:

    parent = self()
    stream = Task.async_stream(
      enum,
      fn (item) ->
        Ecto.Adapters.SQL.Sandbox.allow(Repo, parent, self())
        my_func(item, 2) end,
      [ordered: false, max_concurrency: 1]
    )
    results = Enum.to_list(stream)

and without:

    stream = Task.async_stream(
      enum,
      fn (item) ->
        my_func(item, 2) end,
      [ordered: false, max_concurrency: 1]
    )
    results = Enum.to_list(stream)

And again unfortunately all to no noticeable effect: 19:08:14.135 [error] Postgrex.Protocol (#PID<0.658.0>) disconnected: ** (DBConnection.ConnectionError) owner #PID<0.1748.0> exited while client #PID<0.1751.0> is still running with: shutdown

I’ve realized that I misdiagnosed your problem.

When I tried to reproduce the error you’re getting, I realized that I had been thinking of the other error that occurs when a process cannot find a checked out connection. It looks like this:

11:14:40.287 [error] Task #PID<0.244.0> started from #PID<0.241.0> terminating                                                                                              
** (DBConnection.OwnershipError) cannot find ownership process for #PID<0.244.0>.                                                                                           
                                                                                                                                                                            
When using ownership, you must manage connections in one                                                                                                                    
of the four ways:                                                                                                                                                           

[...it goes on for a while...]

However, the error you’re getting must be caused by the test process exiting before another process stopped using the DB connection. I was only able to reproduce this as follows:

  test "Task.Supervisor.async_stream" do
    enum = [1, 2, 3]

    stream =
      Task.Supervisor.async_stream_nolink(
        Module.TaskSupervisor,
        enum,
        fn item -> my_func(item, 2) end,
        ordered: false,
        max_concurrency: 1
      )

    _result = Enum.to_list(stream)

    # We have to sleep here to allow the process spawned by my_func() 
    # to enter a transaction
    on_exit(fn ->
      Process.sleep(100)
    end)
  end

  defp my_func(user_id, _n) do
    Project.Repo.get(Project.User, user_id)

    # I could only get the error to occur when I spawned another process here.
    spawn(fn ->
      # This transaction starts before the test process exits but it sleeps for a while
      # to let the test process terminate. Then, when Project.Repo.get() is called, it'll
      # raise the error
      # Postgrex.Protocol (#PID<0.210.0>) disconnected: ** (DBConnection.ConnectionError) owner #PID<0.241.0> exited
      Project.Repo.transaction(fn ->
        Process.sleep(100)
        Project.Repo.get(Project.User, user_id)
      end)
    end)
  end

The full error I’m getting looks a bit different from yours, but I think it happen in a similar circumstance:

11:19:10.365 [error] Postgrex.Protocol (#PID<0.210.0>) disconnected: ** (DBConnection.ConnectionError) owner #PID<0.241.0> exited                                           
                                                                                                                                                                            
Client #PID<0.247.0> is still using a connection from owner at location:                                                                                                    
                                                                                                                                                                            
    (elixir) lib/process.ex:228: Process.sleep/1                                                                                                                            
    test/sandbox_test.exs:29: anonymous fn/1 in SandboxTest.my_func/2                                                                                                       
    (ecto_sql) lib/ecto/adapters/sql.ex:814: anonymous fn/3 in Ecto.Adapters.SQL.checkout_or_transaction/4                                                                  
    (db_connection) lib/db_connection.ex:1349: DBConnection.run_transaction/4                                                                                               
                                                                                                                                                                            
The connection itself was checked out by #PID<0.247.0> at location: 

In the end, you were right about your tests running with async: false since that the default, so my analysis was wrong in that regard. However, I think there’s something else going on in your test suite outside of the code snippets you’ve shared so far.