assign_async in live_component (rendered in layout) seems to trigger DBConnection.OwnershipError while testing live view

[noozo]

Environment

• Elixir version (elixir -v): 1.17
• Phoenix version (mix deps): 1.7.14
• Phoenix LiveView version (mix deps): 0.20.17
• Operating system: OSX (as well as ubuntu in github actions)

Overview of the problem

We have a layout that renders a live_component for a sidebar of navigation items. We do an assign_async to be able to query for extra data of these navigation items while the page is already ready to the user (mainly to populate some badges with various data counts on these items).

The test that fails renders that live view and submits a form to create a row in the database. It then fails the test when checking if that object has been persisted.

Relevant code

sidebar_navigation live component (rendered in layout)

  @impl true
  def update(assigns, socket) do
    {:ok,
     socket
     |> assign(assigns)
     |> assign(:badges, %{})
     |> assign_async(:loaded_badges, fn ->
       {:ok, %{loaded_badges: load_badges()}}
     end)}
  end

  attr :current_user, User, required: true

  @impl true
  def render(assigns) do
    ~H"""
    <div id="sidebar-navigation" class="h-screen bg-yellow-grey-50">
      <.async_result :let={badges} assign={@loaded_badges}>
        <:loading>
          <.render_badges badges={%{}} current_user={@current_user} socket={@socket} />
        </:loading>
        <:failed :let={_reason}>there was an error loading the badges information</:failed>
        <.render_badges badges={badges} current_user={@current_user} socket={@socket} />
      </.async_result>
    </div>
    """
  end

test_helper

ExUnit.configure(formatters: [ExUnitFormatter, ExUnitNotifier], capture_log: true)
ExUnit.start()
Ecto.Adapters.SQL.Sandbox.mode(PassionFruit.Repo, :manual)

conn_case (which the test uses)

  setup tags do
    PassionFruit.DataCase.setup_sandbox(tags)
    {:ok, conn: Phoenix.ConnTest.build_conn()}
  end

data_case (of which the conn_case uses the setup_sandbox function)

  def setup_sandbox(tags) do
    pid = Sandbox.start_owner!(PassionFruit.Repo, shared: not tags[:async])

    on_exit(fn ->
      Sandbox.stop_owner(pid)
    end)
  end

markets/new live view test

defmodule PassionFruitWeb.Live.Admin.Markets.NewTest do
  use PassionFruitWeb.ConnCase, async: false

  import Phoenix.LiveViewTest

  alias PassionFruit.Contexts.Markets
  alias PassionFruitWeb.Live.Admin.Markets.Edit

  @endpoint PassionFruitWeb.Endpoint

  describe "Mounting" do
    setup :register_and_log_in_superadmin

    test "connected mount", %{conn: conn} do
      {:ok, _view, html} = live(conn, "/admin/markets/new")
      {:ok, doc} = Floki.parse_document(html)
      [_result] = Floki.find(doc, "h3:fl-contains('Create market')")
    end

    test "can create a new market, by providing a title", %{conn: conn} do
      {:ok, view, _html} = live(conn, "/admin/markets/new")

      result =
        view
        |> element("#create-market form")
        |> render_submit(%{market: %{title: "MK"}})

      # A market should have been created
      assert [market] = Markets.all() <<<<<<<<<<<----- FAILS HERE

      # We should have been redirected to the edit page
      {:error, {:live_redirect, %{to: url}}} = result
      assert url == Routes.live_path(conn, Edit, market.id)
    end
  end
end

The error

     ** (DBConnection.OwnershipError) cannot find ownership process for #PID<0.1103.0>.

     When using ownership, you must manage connections in one
     of the four ways:

     * By explicitly checking out a connection
     * By explicitly allowing a spawned process
     * By running the pool in shared mode
     * By using :caller option with allowed process

     The first two options require every new process to explicitly
     check a connection out or be allowed by calling checkout or
     allow respectively.

     The third option requires a {:shared, pid} mode to be set.
     If using shared mode in tests, make sure your tests are not
     async.

     The fourth option requires [caller: pid] to be used when
     checking out a connection from the pool. The caller process
     should already be allowed on a connection.

     If you are reading this error, it means you have not done one
     of the steps above or that the owner process has crashed.

     See Ecto.Adapters.SQL.Sandbox docs for more information.
     code: assert [market] = Markets.all()
     stacktrace:
       (ecto_sql 3.11.3) lib/ecto/adapters/sql.ex:1051: Ecto.Adapters.SQL.raise_sql_call_error/1
       (ecto_sql 3.11.3) lib/ecto/adapters/sql.ex:952: Ecto.Adapters.SQL.execute/6
       (ecto 3.11.2) lib/ecto/repo/queryable.ex:232: Ecto.Repo.Queryable.execute/4
       (ecto 3.11.2) lib/ecto/repo/queryable.ex:19: Ecto.Repo.Queryable.all/3
       test/passion_fruit_web/live/admin/markets/new_test.exs:29: (test)

Conclusion

The database call inside the process that gets spawned by assign_async seems to kind of "take over" the query that is done in the test itself, when checking if the object was persisted to the database, because the error occurs at the test level.

Setting the test as async: true or false makes no difference.

When i make the assign_async code be just a plain assign, everything works.

[chrismccord]

Can you verify wither update/2 is being called twice (ie when you submit the form it may be called again if the parent's handle_event happens to re-render the live_component call), so you may need to specifically check if you've already async loaded the badges

[noozo]

It does indeed get called twice.

Also, if i run the test in isolation it passes, but if i run it alongside other tests that also render the same layout and sidebar, then it becomes flaky.

[noozo]

> mix test test/passion_fruit_web/live/admin/markets/new_test.exs test/passion_fruit_web/live/admin/projects/new_test.exs test/passion_fruit_web/live/admin/skills/new_test.exs

17:25:02.716 [info] Not generating runtime.exs because no files changed.
Running ExUnit with seed: 534034, max_cases: 16

.....[error] Postgrex.Protocol (#PID<0.623.0>) disconnected: ** (DBConnection.ConnectionError) client #PID<0.927.0> exited
.
Finished in 0.6 seconds (0.6s async, 0.00s sync)
6 tests, 0 failures
[os_mon] cpu supervisor port (cpu_sup): Erlang has closed
[os_mon] memory supervisor port (memsup): Erlang has closed

 ~/Development/web | st-115_conve..gation_to_lv *111 !11 ?4 .................................................... direnv | 19.4.0 node | 27.0 erlang | 1.17.0 elixir
> mix test test/passion_fruit_web/live/admin/markets/new_test.exs test/passion_fruit_web/live/admin/projects/new_test.exs test/passion_fruit_web/live/admin/skills/new_test.exs

17:25:06.329 [info] Not generating runtime.exs because no files changed.
Running ExUnit with seed: 190280, max_cases: 16

...

  1) test Mounting can create a new skill, by providing a title (PassionFruitWeb.Live.Admin.Skills.NewTest)
     test/passion_fruit_web/live/admin/skills/new_test.exs:20
     ** (DBConnection.OwnershipError) cannot find ownership process for #PID<0.837.0>.

     When using ownership, you must manage connections in one
     of the four ways:

     * By explicitly checking out a connection
     * By explicitly allowing a spawned process
     * By running the pool in shared mode
     * By using :caller option with allowed process

     The first two options require every new process to explicitly
     check a connection out or be allowed by calling checkout or
     allow respectively.

     The third option requires a {:shared, pid} mode to be set.
     If using shared mode in tests, make sure your tests are not
     async.

     The fourth option requires [caller: pid] to be used when
     checking out a connection from the pool. The caller process
     should already be allowed on a connection.

     If you are reading this error, it means you have not done one
     of the steps above or that the owner process has crashed.

     See Ecto.Adapters.SQL.Sandbox docs for more information.
     code: %PassionFruit.DataPage{entries: [_skill]} = Skills.paginate(%{query_str: "Wazzzaaa skill"})
     stacktrace:
       (ecto_sql 3.11.3) lib/ecto/adapters/sql.ex:1051: Ecto.Adapters.SQL.raise_sql_call_error/1
       (ecto_sql 3.11.3) lib/ecto/adapters/sql.ex:952: Ecto.Adapters.SQL.execute/6
       (ecto 3.11.2) lib/ecto/repo/queryable.ex:232: Ecto.Repo.Queryable.execute/4
       (ecto 3.11.2) lib/ecto/repo/queryable.ex:19: Ecto.Repo.Queryable.all/3
       (ecto 3.11.2) lib/ecto/repo/queryable.ex:154: Ecto.Repo.Queryable.one/3
       (scrivener_ecto 2.7.0) lib/scrivener/paginater/ecto/query.ex:56: Scrivener.Paginater.Ecto.Query.total_entries/4
       (scrivener_ecto 2.7.0) lib/scrivener/paginater/ecto/query.ex:17: Scrivener.Paginater.Ecto.Query.paginate/2
       (passion_fruit 0.1.1) lib/passion_fruit/contexts/skills.ex:23: PassionFruit.Contexts.Skills.paginate/2
       test/passion_fruit_web/live/admin/skills/new_test.exs:29: (test)

     The following output was logged:
     [error] Postgrex.Protocol (#PID<0.623.0>) disconnected: ** (DBConnection.ConnectionError) client #PID<0.877.0> exited

..
Finished in 0.6 seconds (0.6s async, 0.00s sync)

Failed:

* PassionFruitWeb.Live.Admin.Skills.NewTest :: test Mounting can create a new skill, by providing a title
   mix test test/passion_fruit_web/live/admin/skills/new_test.exs:20

6 tests, 1 failure
[os_mon] cpu supervisor port (cpu_sup): Erlang has closed
[os_mon] memory supervisor port (memsup): Erlang has closed

[noozo]

line 29 of the new_test is assert [market] = Markets.all()

[noozo]

Can you verify wither update/2 is being called twice (ie when you submit the form it may be called again if the parent's handle_event happens to re-render the live_component call), so you may need to specifically check if you've already async loaded the badges

Are you saying that's what causes the test's query to fail? I might be missing something from your comments :)

[chrismccord]

To be clear, I'm asking if your parent LV is calling the LC update/2 twice, while in a connected mount, in which case you'd be issuing two async tasks => the first on connected mount, and potentially the second when interacting with the form.

[noozo]

When i put an IO.inspect inside the LC's update i see this:

> mix test test/passion_fruit_web/live/admin/markets/new_test.exs
Compiling 17 files (.ex)

18:24:50.386 [info] Generating runtime.exs from files under config/runtime...
Running ExUnit with seed: 386095, max_cases: 16

"sidebar navigation update"
"sidebar navigation update"

I would expect that to be the case. But my question is why does the test then fail to perform a database query when the LC has the assign_async?

Btw, this LC is inside the layout used by the new.ex live view, not directly in the view.

[noozo]

I guess my broader question is: why does the number of async tasks matter in terms of the DB ownership problem?

[noozo]

@chrismccord any idea what might be wrong?

[SteffenDE]

@noozo the number of async tasks should not matter concerning ownership. If you can find the time, it would help us a lot if you could create a runnable single file to reproduce.

This should be a good start: https://github.com/phoenixframework/phoenix_live_view/blob/main/.github/single-file-samples/test.exs

[noozo]

@SteffenDE i would need to add a custom layout to the mix, since this live component is inside one, which means even more time spent on this. Not to be mean or anything, but right now i really don't have time to help here :(

On our code i basically modified it to not use assign_async and that works for now (even though it would be nicer to do those queries async).

[voughtdq]

Maybe I missed something, but if you're using manual sandbox checkouts, then that task process would need to call Ecto.Adapters.SQL.Sandbox.allow/3 on itself.

parent = self()
func = fn -> Ecto.Adapters.SQL.Sandbox.allow(Repo, parent, self()) end

at the top of the test case and passing func down into the assign_async and calling it before the query might fix it? At least we'd know if the task PID not being an allowed process is what is causing it.

If that's what's causing it, we can try to find a more graceful method of getting the task PID/allowing it in the sandbox.

[noozo]

I'm following the way normal phoenix apps do it (code generated by mix phx_new):

def setup_sandbox(tags) do
    pid = Sandbox.start_owner!(PassionFruit.Repo, shared: not tags[:async])

    on_exit(fn ->
      Sandbox.stop_owner(pid)
    end)
  end

[voughtdq]

Yes. If you have async: true, the sandbox runs in manual mode, requiring you to manually call Ecto.Adapters.SQL.Sandbox.allow(Repo, parent, self()).

If you have async: false, it is supposed to run in shared mode, but this relies on the process dict being correctly populated. If, at any point, $callers in the process dict gets overwritten or modified (which any call to any library can do if it modifies the process dict), it will cause that DBConnection.OwnershipError if the caller it is expecting isn't there. Just as an example, $callers gets modified here. That's not to say that's causing your problem, but it is possible.

I hypothesize that you're getting the same error, but for different reasons. For async: true, it's because you're not manually allowing the task inside the async_assign. For async: false, I believe it could be caused by $callers getting "corrupted".

To test this, I'm suggesting that you check to see if giving the task an allowance to the sandbox solves the problem. You would run this test with async: true.

Sorry, I know this is kind of invasive, but you will need a few things:

A mount:

defmodule TestCallbackMount do
  def on_mount(:test_callback, _params, _session, socket) do
    test_callback = socket.private.connect_info.assigns[:test_callback]
    {:cont, Phoenix.Component.assign(socket, :test_callback, test_callback)}
  end
end

You will need to wrap the route for the failing test in a live_session:

live_session :default, on_mount: [{TestCallbackMount, :test_callback}] do
  live "/admin/marks/new", Admin.Markets.Index, :new
end

In your actual test case you'll want to assign the callback to your conn:

  require Logger
  defp put_test_callback(conn, callback) when is_function(callback) do
    assign(conn, :test_callback, callback)
  end

  test "...", %{conn: conn} do
    caller = self()
    conn = put_test_callback(conn, fn context -> 
      result = Ecto.Adapters.SQL.Sandbox.allow(PassionFruit.Repo, caller, self())
      Logger.warning(fn -> "sandbox allow result for #{context}: #{inspect(result)}" end)
    end)

   {:ok, view, _html} = live(conn, "/admin/markets/new")
    ...
  end 

Now you can use the callbacks in your live view:

  @impl true
  def mount(_params, _session, socket) do
    if socket.assigns.test_callback do
      socket.assigns.test_callback.(:mount)
    else
      Logger.warning(fn -> "No test callback provided." end)
    end

    {:ok, socket}
  end

and I think it should work in your component too:

  @impl true
  def update(assigns, socket) do
    callback =
      if socket.assigns.test_callback do
        socket.assigns.test_callback
      else
        fn _context -> :noop end
      end

    {:ok,
     socket
     |> assign(assigns)
     |> assign(:badges, %{})
     |> assign_async(:loaded_badges, fn ->
       callback.(:async_loaded_badges)
       {:ok, %{loaded_badges: load_badges()}}
     end)}
  end

This is what the logger results look like in my own local test:

23:02:29.618 request_id=F-KSYDlEzo4f7bwAAAEh [warning] Sandbox allow result mount: {:already, :allowed}
23:02:29.703 [warning] Sandbox allow result mount: :ok
23:02:29.703 [warning] Sandbox allow result assign_async: :ok
23:02:29.740 [warning] Sandbox allow result assign_async: :ok

If this resolves the issue when async: true, then something is wrong with how the sandbox works in shared mode.

[noozo]

@voughtdq wow, thanks for that. will give it a spin to test it out <3

[SteffenDE]

@voughtdq manually calling allow is only needed if the sandbox is in auto mode (which is normally not used). The behavior you're describing with the callers is actually what's used in manual mode. In shared mode, the callers are ignored as everyone uses the same connection.

[noozo]

@SteffenDE in that case, do you have any idea why my problem occurs? Is it specific to a live_component directly inside a layout, for instance?

[SteffenDE]

@noozo sadly not. I tried to reproduce it, but this works fine:

Application.put_env(:phoenix, Example.Endpoint,
  http: [ip: {127, 0, 0, 1}, port: 5001],
  server: true,
  live_view: [signing_salt: "aaaaaaaa"],
  secret_key_base: String.duplicate("a", 64)
)

Application.put_env(:phoenix, Repo,
  database: "assign_async_sample",
  username: "postgres",
  password: "postgres",
  pool: Ecto.Adapters.SQL.Sandbox
)

Mix.install([
  {:plug_cowboy, "~> 2.5"},
  {:jason, "~> 1.0"},
  {:phoenix, "~> 1.7"},
  # please test your issue using the latest version of LV from GitHub!
  {:phoenix_live_view,
   github: "phoenixframework/phoenix_live_view", branch: "main", override: true},
  {:ecto_sql, "~> 3.11"},
  {:postgrex, ">= 0.0.0"},
  {:floki, ">= 0.30.0"}
])

ExUnit.start()

# build the LiveView JavaScript assets (this needs mix and npm available in your path!)
path = Phoenix.LiveView.__info__(:compile)[:source] |> Path.dirname() |> Path.join("../")
System.cmd("mix", ["deps.get"], cd: path, into: IO.binstream())
System.cmd("npm", ["install"], cd: Path.join(path, "./assets"), into: IO.binstream())
System.cmd("mix", ["assets.build"], cd: path, into: IO.binstream())

defmodule Repo do
  use Ecto.Repo,
    adapter: Ecto.Adapters.Postgres,
    otp_app: :phoenix
end

defmodule Migration0 do
  use Ecto.Migration

  def change do
    create table("posts") do
      add(:title, :string)
      timestamps(type: :utc_datetime_usec)
    end

    create table("comments") do
      add(:content, :string)
      add(:post_id, references(:posts, on_delete: :delete_all), null: false)
    end
  end
end

defmodule Post do
  use Ecto.Schema

  schema "posts" do
    field(:title, :string)
    timestamps(type: :utc_datetime_usec)
    has_many(:comments, Comment)
  end
end

defmodule Comment do
  use Ecto.Schema

  schema "comments" do
    field(:content, :string)
    belongs_to(:post, Post)
  end
end

defmodule Example.ErrorView do
  def render(template, _), do: Phoenix.Controller.status_message_from_template(template)
end

defmodule Example.HomeLive do
  use Phoenix.LiveView, layout: {__MODULE__, :live}

  def mount(_params, _session, socket) do
    {:ok, assign(socket, :count, 0)}
  end

  def render("live.html", assigns) do
    ~H"""
    <script src="/assets/phoenix/phoenix.js">
    </script>
    <script src="/assets/phoenix_live_view/phoenix_live_view.js">
    </script>
    <%!-- uncomment to use enable tailwind --%>
    <%!-- <script src="https://cdn.tailwindcss.com"></script> --%>
    <script>
      let liveSocket = new window.LiveView.LiveSocket("/live", window.Phoenix.Socket)
      liveSocket.connect()
    </script>
    <style>
      * { font-size: 1.1em; }
    </style>
    <%= @inner_content %>

    <.live_component id="intemplate" module={Example.LiveComponent} />
    """
  end

  def render(assigns) do
    ~H"""
    <%= @count %>
    <button phx-click="inc">+</button>
    <button phx-click="dec">-</button>
    """
  end

  def handle_event("inc", _params, socket) do
    {:noreply, assign(socket, :count, socket.assigns.count + 1)}
  end

  def handle_event("dec", _params, socket) do
    {:noreply, assign(socket, :count, socket.assigns.count - 1)}
  end
end

defmodule Example.LiveComponent do
  use Phoenix.LiveComponent

  def update(_assigns, socket) do
    {:ok, assign_async(socket, :posts, fn -> {:ok, %{posts: Repo.all(Post)}} end)}
  end

  def handle_event("create", _, socket) do
    Repo.insert!(%Post{title: "Post 3", comments: [%Comment{content: "Comment 4"}]})

    {:noreply, assign_async(socket, :posts, fn -> {:ok, %{posts: Repo.all(Post)}} end)}
  end

  def render(assigns) do
    ~H"""
    <div>
      <button phx-click="create" phx-target={@myself}>Create Post</button>
      <%= inspect(@posts) %>
    </div>
    """
  end
end

defmodule Example.Router do
  use Phoenix.Router
  import Phoenix.LiveView.Router

  pipeline :browser do
    plug(:accepts, ["html"])
  end

  scope "/", Example do
    pipe_through(:browser)

    live("/", HomeLive, :index)
  end
end

defmodule Example.Endpoint do
  use Phoenix.Endpoint, otp_app: :phoenix
  socket("/live", Phoenix.LiveView.Socket)

  plug Plug.Static, from: {:phoenix, "priv/static"}, at: "/assets/phoenix"
  plug Plug.Static, from: {:phoenix_live_view, "priv/static"}, at: "/assets/phoenix_live_view"

  plug(Example.Router)
end

defmodule Main do
  import Ecto.Query, warn: false

  def main do
    _ = Repo.__adapter__().storage_down(Repo.config())
    :ok = Repo.__adapter__().storage_up(Repo.config())

    {:ok, _} = Supervisor.start_link([Repo, Example.Endpoint], strategy: :one_for_one)
    Ecto.Migrator.run(Repo, [{0, Migration0}], :up, all: true, log_migrations_sql: :debug)

    Repo.insert!(%Post{
      title: "Post 1",
      comments: [%Comment{content: "Comment 1"}, %Comment{content: "Comment 2"}]
    })

    Repo.insert!(%Post{title: "Post 2", comments: [%Comment{content: "Comment 3"}]})
    ExUnit.run()
    Process.sleep(:infinity)
  end
end

defmodule Example.HomeLiveTest do
  use ExUnit.Case, async: true

  import Phoenix.ConnTest
  import Plug.Conn
  import Phoenix.LiveViewTest

  @endpoint Example.Endpoint

  alias Ecto.Adapters.SQL.Sandbox

  def setup_sandbox(tags) do
    pid = Sandbox.start_owner!(Repo, shared: not tags[:async])

    on_exit(fn ->
      Sandbox.stop_owner(pid)
    end)
  end

  setup :setup_sandbox

  test "works properly" do
    conn = Phoenix.ConnTest.build_conn()

    {:ok, view, _html} = live(conn, "/")

    html = render_async(view)
    assert html =~ "Post 1"
    assert html =~ "Post 2"

    assert Repo.all(Post) |> length() == 2

    view
    |> element("button", "Create")
    |> render_click()

    html = render_async(view)
    assert Repo.all(Post) |> length() == 3

    assert html =~ "Post 1"
    assert html =~ "Post 2"
    assert html =~ "Post 3"
  end
end

Main.main()

[voughtdq]

@SteffenDE

I can get it to fail with an ownership error if Repo.all/1 gets called in another process.

@noozo does load_badges() do something where it calls your repo in another process?

I modified the test to use this function in the async_assign:

  def all_posts do
    self = self()
    Process.spawn(fn ->
      send(self, {:posts, Repo.all(Post)})
    end, [:link])
    receive do
      {:posts, posts} -> posts
    end
  end

The modified test:

Application.put_env(:phoenix, Example.Endpoint,
  http: [ip: {127, 0, 0, 1}, port: 5001],
  server: true,
  live_view: [signing_salt: "aaaaaaaa"],
  secret_key_base: String.duplicate("a", 64)
)

Application.put_env(:phoenix, Repo,
  database: "assign_async_sample",
  username: "postgres",
  password: "postgres",
  pool: Ecto.Adapters.SQL.Sandbox
)

Mix.install([
  {:plug_cowboy, "~> 2.5"},
  {:jason, "~> 1.0"},
  {:phoenix, "~> 1.7"},
  # please test your issue using the latest version of LV from GitHub!
  {:phoenix_live_view,
   github: "phoenixframework/phoenix_live_view", branch: "main", override: true},
  {:ecto_sql, "~> 3.11"},
  {:postgrex, ">= 0.0.0"},
  {:floki, ">= 0.30.0"}
])

ExUnit.start()

# build the LiveView JavaScript assets (this needs mix and npm available in your path!)
path = Phoenix.LiveView.__info__(:compile)[:source] |> Path.dirname() |> Path.join("../")
System.cmd("mix", ["deps.get"], cd: path, into: IO.binstream())
System.cmd("npm", ["install"], cd: Path.join(path, "./assets"), into: IO.binstream())
System.cmd("mix", ["assets.build"], cd: path, into: IO.binstream())

defmodule Repo do
  use Ecto.Repo,
    adapter: Ecto.Adapters.Postgres,
    otp_app: :phoenix
end

defmodule Migration0 do
  use Ecto.Migration

  def change do
    create table("posts") do
      add(:title, :string)
      timestamps(type: :utc_datetime_usec)
    end

    create table("comments") do
      add(:content, :string)
      add(:post_id, references(:posts, on_delete: :delete_all), null: false)
    end
  end
end

defmodule Post do
  use Ecto.Schema

  schema "posts" do
    field(:title, :string)
    timestamps(type: :utc_datetime_usec)
    has_many(:comments, Comment)
  end
end

defmodule Comment do
  use Ecto.Schema

  schema "comments" do
    field(:content, :string)
    belongs_to(:post, Post)
  end
end

defmodule Example.ErrorView do
  def render(template, _), do: Phoenix.Controller.status_message_from_template(template)
end

defmodule Example.HomeLive do
  use Phoenix.LiveView, layout: {__MODULE__, :live}

  def mount(_params, _session, socket) do
    {:ok, assign(socket, :count, 0)}
  end

  def render("live.html", assigns) do
    ~H"""
    <script src="/assets/phoenix/phoenix.js">
    </script>
    <script src="/assets/phoenix_live_view/phoenix_live_view.js">
    </script>
    <%!-- uncomment to use enable tailwind --%>
    <%!-- <script src="https://cdn.tailwindcss.com"></script> --%>
    <script>
      let liveSocket = new window.LiveView.LiveSocket("/live", window.Phoenix.Socket)
      liveSocket.connect()
    </script>
    <style>
      * { font-size: 1.1em; }
    </style>
    <%= @inner_content %>

    <.live_component id="intemplate" module={Example.LiveComponent} />
    """
  end

  def render(assigns) do
    ~H"""
    <%= @count %>
    <button phx-click="inc">+</button>
    <button phx-click="dec">-</button>
    """
  end

  def handle_event("inc", _params, socket) do
    {:noreply, assign(socket, :count, socket.assigns.count + 1)}
  end

  def handle_event("dec", _params, socket) do
    {:noreply, assign(socket, :count, socket.assigns.count - 1)}
  end
end

defmodule Example.LiveComponent do
  use Phoenix.LiveComponent

  def update(_assigns, socket) do
    {:ok, assign_async(socket, :posts, fn -> 
      {:ok, %{posts: all_posts()}} 
    end)}
  end

  def handle_event("create", _, socket) do
    Repo.insert!(%Post{title: "Post 3", comments: [%Comment{content: "Comment 4"}]})

    {:noreply, assign_async(socket, :posts, fn ->
      {:ok, %{posts: all_posts()}} 
    end)}
  end

  def render(assigns) do
    ~H"""
    <div>
      <button phx-click="create" phx-target={@myself}>Create Post</button>
      <%= inspect(@posts) %>
    </div>
    """
  end

  def all_posts do
    self = self()
    Process.spawn(fn ->
      send(self, {:posts, Repo.all(Post)})
    end, [:link])
    receive do
      {:posts, posts} -> posts
    end
  end
end

defmodule Example.Router do
  use Phoenix.Router
  import Phoenix.LiveView.Router

  pipeline :browser do
    plug(:accepts, ["html"])
  end

  scope "/", Example do
    pipe_through(:browser)

    live("/", HomeLive, :index)
  end
end

defmodule Example.Endpoint do
  use Phoenix.Endpoint, otp_app: :phoenix
  socket("/live", Phoenix.LiveView.Socket)

  plug Plug.Static, from: {:phoenix, "priv/static"}, at: "/assets/phoenix"
  plug Plug.Static, from: {:phoenix_live_view, "priv/static"}, at: "/assets/phoenix_live_view"

  plug(Example.Router)
end

defmodule Main do
  import Ecto.Query, warn: false

  def main do
    _ = Repo.__adapter__().storage_down(Repo.config())
    :ok = Repo.__adapter__().storage_up(Repo.config())

    {:ok, _} = Supervisor.start_link([Repo, Example.Endpoint], strategy: :one_for_one)
    Ecto.Migrator.run(Repo, [{0, Migration0}], :up, all: true, log_migrations_sql: :debug)

    Repo.insert!(%Post{
      title: "Post 1",
      comments: [%Comment{content: "Comment 1"}, %Comment{content: "Comment 2"}]
    })

    Repo.insert!(%Post{title: "Post 2", comments: [%Comment{content: "Comment 3"}]})
    Ecto.Adapters.SQL.Sandbox.mode(Repo, :manual)
    ExUnit.run()
    Process.sleep(:infinity)
  end
end

defmodule Example.HomeLiveTest do
  use ExUnit.Case, async: true

  import Phoenix.ConnTest
  import Plug.Conn
  import Phoenix.LiveViewTest

  @endpoint Example.Endpoint

  alias Ecto.Adapters.SQL.Sandbox

  def setup_sandbox(tags) do
    pid = Sandbox.start_owner!(Repo, shared: not tags[:async])

    on_exit(fn ->
      Sandbox.stop_owner(pid)
    end)
  end

  setup :setup_sandbox

  test "works properly" do
    conn = Phoenix.ConnTest.build_conn()

    {:ok, view, _html} = live(conn, "/")

    html = render_async(view)
    assert html =~ "Post 1"
    assert html =~ "Post 2"

    assert Repo.all(Post) |> length() == 2

    view
    |> element("button", "Create")
    |> render_click()

    html = render_async(view)
    assert Repo.all(Post) |> length() == 3

    assert html =~ "Post 1"
    assert html =~ "Post 2"
    assert html =~ "Post 3"
  end
end

Main.main()

[SteffenDE]

@voughtdq there is an important difference and your question is very good.

If the function in assign_async does indeed spawn another process, the ownership error is expected. If you’d use Task for starting the process it should work because of the callers list, but spawn itself would need an explicit allow.

@noozo please show us your Markets.all and load_badges implementation :)

[noozo]

load_badges (and co.):

  defp load_badges do
    %{
      recent_leads: recent_leads(),
      stalled_project_matches: stalled_project_matches(),
      stalled_projects: stalled_projects(),
      payments_that_require_attention: payments_that_require_attention(),
      specialist_profiles_to_verify: specialist_profiles_to_verify(),
      unverified_companies: unverified_companies(),
      unverified_specialist_references: unverified_specialist_references(),
      unpublished_specialist_references: unpublished_specialist_references(),
      oban_jobs_with_errors: oban_jobs_with_errors(),
      roles_without_questions: Enum.count(PassionFruit.Contexts.Roles.without_questions()),
      specialisms_without_roles: Enum.count(Specialisms.without_roles()),
      roles_without_competencies: Enum.count(Roles.without_competencies())
    }
  end

  defp recent_leads, do: Leads.count_recent()

  defp stalled_projects, do: Enum.count(Projects.list_stalled_ids())

  defp payments_that_require_attention,
    do:
      Enum.count(CompanyPayments.failed_last_log_entries_ids()) +
        Enum.count(SpecialistPayments.failed_last_log_entries_ids()) +
        Enum.count(SpecialistPayments.waiting_for_confirmation_ids())

  defp stalled_project_matches, do: Enum.count(ProjectMatches.list_stalled_ids())

  defp specialist_profiles_to_verify, do: Enum.count(SpecialistProfiles.list_ids_to_verify())

  defp unverified_companies, do: Companies.unverified_count()

  defp unverified_specialist_references, do: SpecialistReferences.unverified_count()
  defp unpublished_specialist_references, do: SpecialistReferences.unpublished_count()

  defp oban_jobs_with_errors, do: Enum.count(ObanJobs.jobs_with_errors())

These are all pretty much just plain Ecto, for instance (Leads):

  def count_recent do
    recent_timeframe = Timex.shift(DateTime.utc_now(), weeks: -Lead.recent_lead_in_weeks())

    Repo.aggregate(
      from(l in Lead, where: l.inserted_at >= ^recent_timeframe),
      :count
    )
  end

or (ProjectMatches)

  def list_stalled_ids do
    query =
      from(
        p in ProjectMatch,
        select: p.id
      )

    query
    |> filter_by(:stalled, "true")
    |> Repo.all()
  end

The Markets.all() is also simple:

  def all do
    query =
      from(
        m in Market,
        where: is_nil(m.deleted_at),
        order_by: [asc: m.title]
      )

    Repo.all(query)
  end

[voughtdq]

Are you using caching for any of those calls (e.g. a library like Cachex)?

[noozo]

No, unless Ecto has something like that on by default.

[voughtdq]

Ok wait. You use Oban it seems. Does this

result =
        view
        |> element("#create-market form")
        |> render_submit(%{market: %{title: "MK"}})

potentially trigger an oban job?