ug0
Problem with testing bulk inserts with Task.async_stream
I have code to do bulk insert which looks like this:
stream = build_attrs_stream()
Repo.transaction(fn ->
stream
|> Stream.chunk_every(5000)
|> Task.async_stream(fn batch ->
{n, _} = Repo.insert_all(Assignment, batch)
n
end, ordered: false)
|> Enum.reduce(0, fn {:ok, n}, acc -> acc + n end)
end)
The above code works fine until I run the test, which gives me error:
(EXIT from #PID<0.544.0>) an exception was raised:
** (DBConnection.ConnectionError) connection not available and request was dropped from queue after 939ms. You can configure how long requests wait in the queue using :queue_target and :queue_interval. See DBConnection.start_link/2 for more information
I’m using Phoenix v1.4 and Ecto v3.1, database is Postgres.
Any help?
Most Liked
benwilson512
Hey @ug0 this is because of how the ecto sandbox works during tests. Each test process is provided its own database connection. The Task.async_stream call spawns new processes, and ecto doesn’t know which connection to use. You can solve this by giving Ecto a little more information:
stream = build_attrs_stream()
parent = self()
Repo.transaction(fn ->
stream
|> Stream.chunk_every(5000)
|> Task.async_stream(fn batch ->
{n, _} = Repo.insert_all(Assignment, batch, caller: parent)
n
end, ordered: false)
|> Enum.reduce(0, fn {:ok, n}, acc -> acc + n end)
end)
The change here is the addition of caller: parent as options to the Repo call.
Last Post!
gmile
Sorry for necro-posting. This is how I solved the issue combining Repo.stream and Task.async_stream with in my case.
Below is a self-contained script that will hang:
Mix.install([
{:ecto_sql, "~> 3.10"},
{:postgrex, ">= 0.0.0"}
])
Application.put_env(:myapp, Repo, database: "repo_stream_example", pool: Ecto.Adapters.SQL.Sandbox)
defmodule Repo do
use Ecto.Repo, adapter: Ecto.Adapters.Postgres, otp_app: :myapp
end
defmodule Main do
def my_fun do
import Ecto.Query
Repo.transaction(fn ->
from(n in fragment("SELECT generate_series(1, 3) AS n"), select: n.n)
|> Repo.stream()
|> Task.async_stream(fn number ->
# this hangs
Repo.query!("select $1 + 1", [number]).rows
end)
|> Enum.to_list()
end)
end
def main do
Repo.__adapter__().storage_down(Repo.config())
:ok = Repo.__adapter__().storage_up(Repo.config())
{:ok, _} = Repo.start_link([])
end
end
ExUnit.start()
defmodule MainTest do
use ExUnit.Case
setup do
:ok = Ecto.Adapters.SQL.Sandbox.checkout(Repo)
Ecto.Adapters.SQL.Sandbox.mode(Repo, {:shared, self()})
:ok
end
test "the insides of my_fun hang" do
assert {:ok, [ok: [[2]], ok: [[3]], ok: [[4]]]} = Main.my_fun()
end
end
Main.main()
ExUnit.run()
Make the following change:
diff --git a/hang-repro.exs b/hang-repro.exs
index 3be69f09da..1bf6f76929 100644
--- a/hang-repro.exs
+++ b/hang-repro.exs
@@ -17,7 +17,6 @@ defmodule Main do
from(n in fragment("SELECT generate_series(1, 3) AS n"), select: n.n)
|> Repo.stream()
|> Task.async_stream(fn number ->
- # this hangs
Repo.query!("select $1 + 1", [number]).rows
end)
|> Enum.to_list()
@@ -35,14 +34,8 @@ ExUnit.start()
defmodule MainTest do
use ExUnit.Case
-
- setup do
- :ok = Ecto.Adapters.SQL.Sandbox.checkout(Repo)
- Ecto.Adapters.SQL.Sandbox.mode(Repo, {:shared, self()})
- :ok
- end
- test "the insides of my_fun hang" do
+ test "the insides of my_fun don't hang anymore" do
assert {:ok, [ok: [[2]], ok: [[3]], ok: [[4]]]} = Main.my_fun()
end
end
This way a default value for Sandbox mode is used, which is :auto. The script will now run to completion.
I think it’s important to be mention, however, that the code inside Task.async_stream will not run inside the same transaction, and won’t be automatically rolled back at the end of the test. So any data inserted in the DB during the test will have to be cleaned up manually. But that’s OK in my case.
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









