Oban bringing app down with `Repo not available` error In development

johantell · March 8, 2023, 9:15pm

We’ve come across a somewhat frustrating problem on our development machines where the application gets shut down after Oban fails with the following error:

** (UndefinedFunctionError) function MyApp.Repo.transaction/2 is undefined (module MyApp.Repo is not available)
    (my_app 0.1.0) MyApp.Repo.transaction(#Function<0.22423664/0 in Oban.Stager.check_leadership_and_stage/1>, [log: false, prefix: "public", telemetry_options: [oban_conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}]])

I’ve been successfully able to reproduce (in our) it by:

Starting the phoenix app with Oban
Running mix gettext.extract --merge
Reloading the page

When reloading the page the application stops responding and the full console output is this:

Compiling 424 files (.ex)
[error] GenServer {Oban.Registry, {Oban, Oban.Stager}} terminating
** (UndefinedFunctionError) function MyApp.Repo.transaction/2 is undefined (module MyApp.Repo is not available)
    (my_app 0.1.0) MyApp.Repo.transaction(#Function<0.22423664/0 in Oban.Stager.check_leadership_and_stage/1>, [log: false, prefix: "public", telemetry_options: [oban_conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}]])
    (oban 2.14.2) lib/oban/stager.ex:86: anonymous fn/2 in Oban.Stager.handle_info/2
    (telemetry 0.4.3) /Users/johantell/Projects/my_app_elixir/deps/telemetry/src/telemetry.erl:272: :telemetry.span/3
    (oban 2.14.2) lib/oban/stager.ex:85: Oban.Stager.handle_info/2
    (stdlib 4.2) gen_server.erl:1123: :gen_server.try_dispatch/4
    (stdlib 4.2) gen_server.erl:1200: :gen_server.handle_msg/6
    (stdlib 4.2) proc_lib.erl:240: :proc_lib.init_p_do_apply/3
Last message: :stage
State: %Oban.Stager.State{conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}, name: {:via, Registry, {Oban.Registry, {Oban, Oban.Stager}}}, timer: #Reference<0.3262115946.2793144325.226216>, interval: 1000, limit: 5000, mode: :global, ping_at_tick: 60, swap_at_tick: 65, tick: 40}
[error] GenServer {Oban.Registry, {Oban, Oban.Stager}} terminating
** (UndefinedFunctionError) function MyApp.Repo.query/3 is undefined (module MyApp.Repo is not available)
    (my_app 0.1.0) MyApp.Repo.query("SELECT pg_notify($1, payload) FROM json_array_elements_text($2::json) AS payload", ["public.oban_stager", ["{\"ping\":\"pong\"}"]], [log: false, prefix: "public", telemetry_options: [oban_conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}]])
    (oban 2.14.2) lib/oban/notifiers/postgres.ex:100: Oban.Notifiers.Postgres.notify/3
    (oban 2.14.2) lib/oban/notifier.ex:188: anonymous fn/2 in Oban.Notifier.with_span/4
    (telemetry 0.4.3) /Users/johantell/Projects/my_app_elixir/deps/telemetry/src/telemetry.erl:272: :telemetry.span/3
    (oban 2.14.2) lib/oban/stager.ex:178: Oban.Stager.check_notify_mode/1
    (oban 2.14.2) lib/oban/stager.ex:69: Oban.Stager.handle_continue/2
    (stdlib 4.2) gen_server.erl:1123: :gen_server.try_dispatch/4
    (stdlib 4.2) gen_server.erl:865: :gen_server.loop/7
    (stdlib 4.2) proc_lib.erl:240: :proc_lib.init_p_do_apply/3
Last message: {:continue, :start}
State: %Oban.Stager.State{conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}, name: {:via, Registry, {Oban.Registry, {Oban, Oban.Stager}}}, timer: nil, interval: 1000, limit: 5000, mode: :global, ping_at_tick: 0, swap_at_tick: 5, tick: 0}
[error] GenServer {Oban.Registry, {Oban, Oban.Stager}} terminating
** (UndefinedFunctionError) function MyApp.Repo.query/3 is undefined (module MyApp.Repo is not available)
    (my_app 0.1.0) MyApp.Repo.query("SELECT pg_notify($1, payload) FROM json_array_elements_text($2::json) AS payload", ["public.oban_stager", ["{\"ping\":\"pong\"}"]], [log: false, prefix: "public", telemetry_options: [oban_conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}]])
    (oban 2.14.2) lib/oban/notifiers/postgres.ex:100: Oban.Notifiers.Postgres.notify/3
    (oban 2.14.2) lib/oban/notifier.ex:188: anonymous fn/2 in Oban.Notifier.with_span/4
    (telemetry 0.4.3) /Users/johantell/Projects/my_app_elixir/deps/telemetry/src/telemetry.erl:272: :telemetry.span/3
    (oban 2.14.2) lib/oban/stager.ex:178: Oban.Stager.check_notify_mode/1
    (oban 2.14.2) lib/oban/stager.ex:69: Oban.Stager.handle_continue/2
    (stdlib 4.2) gen_server.erl:1123: :gen_server.try_dispatch/4
    (stdlib 4.2) gen_server.erl:865: :gen_server.loop/7
    (stdlib 4.2) proc_lib.erl:240: :proc_lib.init_p_do_apply/3
Last message: {:continue, :start}
State: %Oban.Stager.State{conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}, name: {:via, Registry, {Oban.Registry, {Oban, Oban.Stager}}}, timer: nil, interval: 1000, limit: 5000, mode: :global, ping_at_tick: 0, swap_at_tick: 5, tick: 0}
[error] GenServer {Oban.Registry, {Oban, Oban.Stager}} terminating
** (UndefinedFunctionError) function MyApp.Repo.query/3 is undefined (module MyApp.Repo is not available)
    (my_app 0.1.0) MyApp.Repo.query("SELECT pg_notify($1, payload) FROM json_array_elements_text($2::json) AS payload", ["public.oban_stager", ["{\"ping\":\"pong\"}"]], [log: false, prefix: "public", telemetry_options: [oban_conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}]])
    (oban 2.14.2) lib/oban/notifiers/postgres.ex:100: Oban.Notifiers.Postgres.notify/3
    (oban 2.14.2) lib/oban/notifier.ex:188: anonymous fn/2 in Oban.Notifier.with_span/4
    (telemetry 0.4.3) /Users/johantell/Projects/my_app_elixir/deps/telemetry/src/telemetry.erl:272: :telemetry.span/3
    (oban 2.14.2) lib/oban/stager.ex:178: Oban.Stager.check_notify_mode/1
    (oban 2.14.2) lib/oban/stager.ex:69: Oban.Stager.handle_continue/2
    (stdlib 4.2) gen_server.erl:1123: :gen_server.try_dispatch/4
    (stdlib 4.2) gen_server.erl:865: :gen_server.loop/7
    (stdlib 4.2) proc_lib.erl:240: :proc_lib.init_p_do_apply/3
Last message: {:continue, :start}
State: %Oban.Stager.State{conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}, name: {:via, Registry, {Oban.Registry, {Oban, Oban.Stager}}}, timer: nil, interval: 1000, limit: 5000, mode: :global, ping_at_tick: 0, swap_at_tick: 5, tick: 0}
[error] GenServer {Oban.Registry, {Oban, Oban.Peer}} terminating
** (UndefinedFunctionError) function MyApp.Repo.transaction/2 is undefined (module MyApp.Repo is not available)
    (my_app 0.1.0) MyApp.Repo.transaction(#Function<1.78375590/0 in Oban.Peers.Postgres.terminate/2>, [log: false, prefix: "public", telemetry_options: [oban_conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}]])
    (oban 2.14.2) lib/oban/peers/postgres.ex:68: Oban.Peers.Postgres.terminate/2
    (stdlib 4.2) gen_server.erl:1161: :gen_server.try_terminate/3
    (stdlib 4.2) gen_server.erl:1351: :gen_server.terminate/10
    (stdlib 4.2) proc_lib.erl:240: :proc_lib.init_p_do_apply/3
Last message: {:EXIT, #PID<0.17657.0>, :shutdown}
State: %Oban.Peers.Postgres.State{conf: %Oban.Config{dispatch_cooldown: 5, engine: Oban.Engines.Basic, get_dynamic_repo: nil, log: false, name: Oban, node: "t@localhost", notifier: Oban.Notifiers.Postgres, peer: Oban.Peers.Postgres, plugins: [{Oban.Plugins.Cron, [crontab: [{"0 9-17 * * *", MyApp.ScheduledJob, [queue: :custify]}]]}, {Oban.Plugins.Pruner, []}], prefix: "public", queues: [default: [limit: 10], mailers: [limit: 20], webhooks: [limit: 5], custify: [limit: 1]], repo: MyApp.Repo, shutdown_grace_period: 15000, stage_interval: 1000, testing: :disabled}, name: {:via, Registry, {Oban.Registry, {Oban, Oban.Peer}}}, timer: #Reference<0.3262115946.2793144323.228688>, interval: 30000, leader?: true, leader_boost: 2}

[notice] Application my_app exited: shutdown

I sadly have to little knowledge about how the code reloader and compiler handles when files are recompiled like this but we only started seeing the problems after upgrading to Oban 2.14.x.

Anyone has experienced this or have any good ideas on how we can try to sort it out?

BradS2S · March 8, 2023, 10:02pm

Do you have MyApp.Repo worker added to the application file?

johantell · March 9, 2023, 7:24am

Yes, everything runs as expected until I run the gettext command (which is forcing the recompilation)

sorentwo · March 10, 2023, 3:46pm

What’s puzzling to me about this issue is how gettext causes MyApp.Repo to recompile, and why does that affect references to MyApp.Repo within Oban? This isn’t an issue I’ve seen reported before, so I have a hunch there’s something peculiar about live reloading in your application.

Hopefully somebody else has some insight!

johantell · March 10, 2023, 8:01pm

From my understanding gettext will force a recompilation of the whole project to extract translations but I would assume that if that were the main source of problem others would have seen it too.

Am I using mix xref wrong when checking for dependencies like this?

$ mix xref graph --source lib/my_app/repo.ex
lib/my_app/repo.ex
└── lib/my_app/repo/soft_delete.ex

$ mix xref graph --source lib/my_app/repo/soft_delete.ex
lib/my_app/repo/soft_delete.ex

I can try to set up a new empty project and see if I manage to reproduce it there as well

BartOtten · March 10, 2023, 10:58pm

Does is also happen when executing ‘mix compile —force’?

johantell · March 13, 2023, 9:37am

Yes I realized now that it does.

I also tried to reproduce it in a newly created app but failed to do so which means I’m back to square one. Any thoughts on how I can investigate deeper? I feel like I lack ways to do so

katafrakt · March 13, 2023, 7:30pm

The output of mix xref looks a bit suspicious to me. What sits in this soft_delete.ex file? Aren’t you, by chance or mistake, redefining MyApp.Repo module inside it?

johantell · March 14, 2023, 8:00am

It’s more or less a copy of GitHub - revelrylabs/ecto_soft_delete: Soft Deletion for Ecto with some extra additions that we haven’t tried to upstream yet.

I don’t think it would have to redefine the Repo but I’m also not an expert on compile time dependencies

There are no secrets in it so I can share it:

defmodule Boardclic.Repo.SoftDelete do
  @moduledoc """
  Manages soft deletes by adding a possibility to jack into ecto's `prepare_query/3` in order
  to modify queries just before they are executed.

  NOTE: This module is an extension of the `Ecto.Repo.SoftDelete` module from
  the `ecto_soft_delete` library that also supports enforcing the soft delete check onto joins.
  """

  import Ecto.Query

  @doc """
  Adds checks to prevent soft deleted records from being returned in our queries.

  It will be able to handle both direct soft deletions (the FROM clause if a single table query)
  and on joined tables.

  To also include soft deleted records, you can pass `include_soft_deleted: true` to your `Repo` call:

  `
  from(o in Organization)
  |> order_by(asc: :name)
  |> Repo.all(include_soft_deleted: true)
  `
  """
  def ignore_soft_deletes(query, opts \\ []) do
    if Keyword.get(opts, :include_soft_deleted, false) do
      query
    else
      query
      |> add_deleted_at_checks_for_base_query()
      |> add_deleted_at_checks_for_joins()
    end
  end

  defp add_deleted_at_checks_for_base_query(query) do
    base_query_module = get_schema_module_from_query(query)

    if soft_deletable?(base_query_module) and !has_include_deleted_at_clause?(query) do
      apply_soft_deletion_check(query, 0)
    else
      query
    end
  end

  defp add_deleted_at_checks_for_joins(query) do
    query.joins
    |> Enum.filter(&soft_deletable_join?(&1, query))
    |> Enum.reduce(query, fn join, query ->
      join_position = Enum.find_index(query.joins, &(&1 == join)) + 1

      apply_soft_deletion_check(query, join_position)
    end)
  end

  # Checks the query to see if it contains a where not is_nil(deleted_at)
  # if it does, we want to be sure that we don't exclude soft deleted records
  defp has_include_deleted_at_clause?(%Ecto.Query{wheres: wheres}) do
    Enum.any?(wheres, fn %{expr: expr} ->
      expr == {:not, [], [{:is_nil, [], [{{:., [], [{:&, [], [0]}, :deleted_at]}, [], []}]}]}
    end)
  end

  def soft_deletable_join?(%Ecto.Query.JoinExpr{source: {_table_name, schema}}, _quert) do
    soft_deletable?(schema)
  end

  def soft_deletable_join?(%Ecto.Query.JoinExpr{assoc: {_index, assoc}}, query) do
    query
    |> assoc_module(assoc)
    |> soft_deletable?()
  end

  def soft_deletable_join?(_join, _query), do: false

  defp soft_deletable?(schema_module) when is_nil(schema_module), do: false

  defp soft_deletable?(schema_module) do
    Enum.member?(schema_module.__schema__(:fields), :deleted_at)
  end

  defp assoc_module(query, assoc) do
    with schema when not is_nil(schema) <- get_schema_module_from_query(query) do
      case Ecto.Association.association_from_schema!(schema, assoc) do
        %{related: related} ->
          related

        # Has through operations will be handled by direct preloads but can at the moment not
        # handle in query joins like `from u in User, join: assoc(u, :organization_user)`
        %Ecto.Association.HasThrough{} ->
          nil

        nil ->
          nil
      end
    end
  end

  defp apply_soft_deletion_check(query, 0), do: from(r in query, where: is_nil(r.deleted_at))
  defp apply_soft_deletion_check(query, 1), do: from([a, r] in query, where: is_nil(r.deleted_at))
  defp apply_soft_deletion_check(query, 2), do: from([a, b, r] in query, where: is_nil(r.deleted_at))
  defp apply_soft_deletion_check(query, 3), do: from([a, b, c, r] in query, where: is_nil(r.deleted_at))

  defp get_schema_module_from_query(%Ecto.Query{from: %{source: {_name, module}}}), do: module
  defp get_schema_module_from_query(_), do: nil
end

katafrakt · March 14, 2023, 9:25am

Oh, and you have use Boardclic.Repo.SoftDelete in Boardclic.Repo - that makes sense now why xref shows what it shows. So, any chance you actually have MyApp.Repo instead of Boardclic.Repo somewhere in the Oban config?

johantell · March 15, 2023, 4:04pm

Right, sorry i replaced all references to “boardclic” in the other code I wrote. We only have one repository set up so there is not any faulty references in the configuration.

We’re actually only using an alias to SoftDelete from the Repo module so that shouldn’t be a problem, right?

defmodule Boardclic.Repo do
  use Ecto.Repo,
    otp_app: :boardclic,
    adapter: Ecto.Adapters.Postgres

  alias Boardclic.Repo.SoftDelete

  @impl Ecto.Repo
  def prepare_query(_operation, query, opts) do
    query =
      if opts[:schema_migration] do
        query
      else
        SoftDelete.ignore_soft_deletes(query, opts)
      end

    {query, opts}
  end
end

stevenpdq · February 1, 2024, 9:25am

Did you find a solution for this? We’re getting exactly the same error & crash after anything is recompiled.

johantell · February 1, 2024, 11:13am

Sadly no solution. We don’t see it as often now but it still happens from time to time. Haven’t gotten further on theories either sadly