TELunus
problem asynchronizing ecto calls
I seem to be having an issue making some code asynchronous. I’m running
stream = Task.Supervisor.async_stream_nolink(
Module.TaskSupervisor,
enum,
fn (item) -> my_func(item, 2) end,
[ordered: false, max_concurrency: 1]
)
results = Enum.to_list(stream)
which gives me 12:31:22.290 [error] Postgrex.Protocol (#PID<0.678.0>) disconnected: ** (DBConnection.ConnectionError) owner #PID<0.1768.0> exited while client #PID<0.1771.0> is still running with: shutdown
part of the code that’s run by my_func is an ecto call
IO.puts("before")
Repo.get(schema, id)
IO.puts("after")
and I’m seeing the "before" get printed right before the error message, but I’m not seeing the "after"
On the other hand if I change my code to run synchronously as
stream = Enum.map(enum, fn (item) -> my_func(item, 2) end)
results = Enum.to_list(stream)
then everything works fine. There’s no error message, and both print statements get run.
Any thoughts on what I could be doing wrong or how I can fix this?
Most Liked
alco
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
alco
Just wanted to comment on this. The team working on Elixir does their best to keep Elixir versions backwards-compatible. The worst thing you should get after upgrading to a newer version is a bunch of new deprecation warnings.
I would actually encourage you to upgrade early because if something does get deprecated, you can stop doing that in your code and reduce the amount of work that you would otherwise have to do if you decided to upgrade months later. Dependencies also keep working as before as a rule, but since you may start getting new warnings, that can be something that prompts you to look into gradually upgrading your dependencies as well sooner rather than later. This is again going to be easier than skipping a few version of Elixir and getting a lot more issues with dependencies when you finally decide to upgrade.
TELunus
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.
Last Post!
TELunus
It does seem to be an issue in our test setup, as it looks like you were suggesting, since if I run the same code outside of the test environment it returns correctly, and the message from IO.puts("after") following the call to the Ecto.Repo.get callback shows up.
I suspect you’re right that the issue is outside of what I’ve shared, but I’m uncertain where to look next.
The test file looks roughly (modified to preserve confidentiality) like:
defmodule Project.ModuleTest do
use Project.ModelCase
alias Project.{Module, Factory, TestHelpers}
@moduletag :module_tag
setup do
variables = values
%{
more_variables: more values,
cleanup_func: cleanup_func
} = TestHelpers.setup_stuff_we_need(variable, other_variable)
on_exit(cleanup_func)
built_variables = Factory.build(variables)
%{
names: some_variables
}
end
describe "#description" do
test "test description", %{
names: some_variables
} do
assert {:ok, %{
specific_thing_we_need_to_check: false,
id: result_id
}} = Module.function(thing_id, and, some, values)
# wait for a different second so records that only have 1 second resolution don't collide
Process.sleep(2000)
assert {:ok, %{
specific_thing_we_need_to_check: true,
something: ^specific_value,
thing_id: ^thing_id
}} = Module.function_that_calls_my_func(thing_id, and, some, values) # <--- the line that actually creates the issue
end
end
end
And the file that defines Project.ModelCase looks like
defmodule Project.ModelCase do
use ExUnit.CaseTemplate
using do
quote do
alias Project.Repo
import Ecto
import Ecto.Query, only: [from: 2]
import Project.ModelCase
end
end
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
def errors_on(changeset) do
Ecto.Changeset.traverse_errors(changeset, fn {message, opts} ->
Enum.reduce(opts, message, fn {key, value}, acc ->
String.replace(acc, "%{#{key}}", to_string(value))
end)
end)
end
end
We’ve also got some stuff in config/test.exs, and taking some guesses as to what might be relevant we’ve got
use Mix.Config
config :project, Project.Repo
adapter: Ecto.Adapters.Postgres,
username: System.get_env("TEST_USERNAME"),
password: System.get_env("TEST_PASSWORD") ,
database: System.get_env("TEST_DATABASE"),
pool_size: 3,
pool: Ecto.Adapters.SQL.Sandbox,
types: Project.PostgresTypes,
ownership_timeout: 10 * 60 * 1000
Popular in Questions
Other popular topics
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #elixirconf-us
- #advent-of-code
- #distillery
- #processes
- #api
- #forms
- #metaprogramming
- #security
- #hex










