I have a test in my Phoenix app that is testing a Phoenix.PubSub subscriber that uses Genserver. The subscriber does some database work as part of its handle_info/2. It looks like this:

defmodule MyApp.Accounts.Subscriber do
  use GenServer
  require Logger
  alias Phoenix.PubSub

  alias MyApp.ReferralCode

 # genserver stuff...

  def init(_) do
    PubSub.subscribe(MyApp.PubSub, "accounts")
    {:ok, %{}}
  end

  def handle_call(:get, _, state) do
    {:reply, state, state}
  end

  def handle_info({:register, user}, state) do
    Logger.info("user registered: #{user.name}")

    %ReferralCode{user_id: user.id}
    |> ReferralCode.changeset(%{code: ReferralCode.generate_code(user.name)})
    |> MyApp.Repo.insert!()

    {:noreply, state}
  end
end

I start my subscriber in my app’s application.ex

children = [
  # Start the PubSub system
  {Phoenix.PubSub, name: MyApp.PubSub},
  MyApp.Accounts.Subscriber,
  # Start the Ecto repository
  supervisor(MyApp.Repo, []),
  # Start the endpoint when the application starts
  supervisor(MyAppWeb.Endpoint, [])
]

opts = [strategy: :one_for_one, name: MyApp.Supervisor]

Supervisor.start_link(children, opts)

My test looks like this:

test "sending creating a referral code upon user registration" do
  start_supervised(MyApp.Accounts.Subscriber)
  user = insert(:user)

  Phoenix.PubSub.broadcast(MappApp.PubSub, "accounts", {:register, user})

  assert_eventually(Repo.exists?(ReferralCode))

  stop_supervised(MyApp.Accounts.Subscriber)
end

The test runs fine when run by itself but when the whole suite is run I get the following error:

[error] GenServer MyApp.Accounts.Subscriber terminating
** (stop) exited in: DBConnection.Holder.checkout(#PID<0.970.0>, [log: #Function<9.124843621/1 in Ecto.Adapters.SQL.with_log/3>, cache_statement: "ecto_insert_referral_codes", timeout: 15000, pool_size: 10, pool: DBConnection.Ownership])
    ** (EXIT) shutdown: "owner #PID<0.969.0> exited"
    <stacktrace...>

This looks like it’s an issue with the database connection still being open when the process is terminated so it doesn’t die gracefully. But I’m not sure how to deal with this.

As an aside, I tried running the test without the call to start_supervised and stop_supervised and strangely it still passed. This did not solve the error mentioned above though. Is there an explanation for this?

Side question: Is this a valid way to implement a pubsub system to handle side effects based on events?