assert_enqueued/2 fails to match a job by state=available and scheduled_at

[smaximov]

Environment

• Oban Version: v2.14.2
• PostgreSQL Version: v13
• Elixir & Erlang/OTP Versions (elixir --version): Elixir 1.14.3 (compiled with Erlang/OTP 25)

Current Behavior

Consider these tests:

defmodule Worker do
  use Oban.Worker

  @impl Oban.Worker
  def perform(_job), do: :ok
end

defmodule Test do
  use ExUnit.Case, async: true
  use Oban.Testing, repo: Repo

  alias Ecto.Adapters.SQL.Sandbox

  describe "inserting jobs to execute immediately" do
    # 1
    test "matches enqueued job by both state & scheduled_at" do
      now = DateTime.utc_now()
      Oban.insert!(Worker.new(%{}))

      assert_enqueued(worker: Worker, state: "available", scheduled_at: {now, delta: 5})
    end

    # 2
    test "matches enqueued job by state only" do
      Oban.insert!(Worker.new(%{}))

      assert_enqueued(worker: Worker, state: "available")
    end

    # 3
    test "matches enqueued job by scheduled_at only" do
      now = DateTime.utc_now()
      Oban.insert!(Worker.new(%{}))

      assert_enqueued(worker: Worker, scheduled_at: {now, delta: 5})
    end
  end

  describe "scheduling jobs in the future" do
    # 4
    test "matches delayed job by state=scheduled" do
      now = DateTime.utc_now()
      tomorrow = DateTime.add(now, 1, :day)
      Oban.insert!(Worker.new(%{}, schedule_in: {1, :day}))

      assert_enqueued(worker: Worker, state: "scheduled", scheduled_at: {tomorrow, delta: 5})
    end
  end
end

Click to see the whole script

#!/usr/bin/env elixir

Mix.install([
  :ecto_sql,
  :jason,
  :postgrex,
  :oban
])

Application.put_env(:my_app, Repo,
  socket_dir: "/run/postgresql",
  database: "my_app",
  show_sensitive_data_on_connection_error: true
)

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

defmodule Migration do
  use Ecto.Migration

  def up do
    Oban.Migration.up()
  end
end

case Repo.__adapter__().storage_up(Repo.config()) do
  :ok -> :ok
  {:error, :already_up} -> :ok
  {:error, error} when is_binary(error) -> raise "can't setup database: #{error}"
  {:error, error} -> raise "can't setup datebase: #{inspect(error)}"
end

children = [
  Repo,
  {Oban, repo: Repo, plugins: [Oban.Plugins.Pruner], queues: [default: 10]}
]

{:ok, pid} = Supervisor.start_link(children, strategy: :one_for_one)

Ecto.Migrator.run(Repo, [{0, Migration}], :up, all: true)

ExUnit.start()

defmodule Worker do
  use Oban.Worker

  @impl Oban.Worker
  def perform(_job), do: :ok
end

defmodule Test do
  use ExUnit.Case, async: true
  use Oban.Testing, repo: Repo

  alias Ecto.Adapters.SQL.Sandbox

  setup do
    repo_config =
      Application.fetch_env!(:my_app, Repo) |> Keyword.put(:pool, Ecto.Adapters.SQL.Sandbox)

    start_link_supervised!({Repo, repo_config})

    start_link_supervised!(
      {Oban, repo: Repo, plugins: [Oban.Plugins.Pruner], queues: [default: 10]}
    )

    pid = Sandbox.start_owner!(Repo, shared: false)
    on_exit(fn -> Sandbox.stop_owner(pid) end)
  end

  describe "inserting jobs to execute immediately" do
    # 1
    test "matches enqueued job by both state & scheduled_at" do
      now = DateTime.utc_now()
      Oban.insert!(Worker.new(%{}))

      assert_enqueued(worker: Worker, state: "available", scheduled_at: {now, delta: 5})
    end

    # 2
    test "matches enqueued job by state only" do
      Oban.insert!(Worker.new(%{}))

      assert_enqueued(worker: Worker, state: "available")
    end

    # 3
    test "matches enqueued job by scheduled_at only" do
      now = DateTime.utc_now()
      Oban.insert!(Worker.new(%{}))

      assert_enqueued(worker: Worker, scheduled_at: {now, delta: 5})
    end
  end

  describe "scheduling jobs in the future" do
    # 4
    test "matches delayed job by state=scheduled" do
      now = DateTime.utc_now()
      tomorrow = DateTime.add(now, 1, :day)
      Oban.insert!(Worker.new(%{}, schedule_in: {1, :day}))

      assert_enqueued(worker: Worker, state: "scheduled", scheduled_at: {tomorrow, delta: 5})
    end
  end
end

When I run them, test #1 fails with a confusing message:

  1) test inserting jobs to execute immediately matches enqueued job by both state & scheduled_at (Test)
     test.exs:77
     Expected a job matching:

     %{
       scheduled_at: {~U[2023-03-22 20:50:12.926871Z], [delta: 5]},
       state: "available",
       worker: Worker
     }

     to be enqueued in the "public" schema. Instead found:

     [
       %{
         scheduled_at: ~U[2023-03-22 20:50:12.926792Z],
         state: "available",
         worker: "Worker"
       }
     ]

     code: assert_enqueued(worker: Worker, state: "available", scheduled_at: {now, delta: 5})
     stacktrace:
       test.exs:81: (test)

The job is clearly there but assert_enqueued/2 fails to match it.

Tests #2 & #3 successfully match this job by state and by scheduled_at, separately. Interestingly enough, test #4 also succeeds by matching a scheduled/delayed job by both state and scheduled_at, so I suppose the issue has something to do with state available.

Expected Behavior

All tests should pass.

[smaximov]

So I've found out that using both state and scheduled_at options forces assert_enqueued/2 to generate the following query:

#Ecto.Query<from j0 in Oban.Job, where: j0.state in ["available", "scheduled"],
 where: fragment(
  "? BETWEEN ? AND ?",
  j0.scheduled_at,
  ^~U[2023-03-22 21:36:32.819404Z],
  ^~U[2023-03-22 21:36:42.819404Z]
),
 where: j0.state == ^"scheduled", where: j0.worker == ^"Worker",
 order_by: [desc: j0.id]>

with an invalid condition state == ^"scheduled". This happens because assert_enqueued/2 builds a Job changeset to process/normalize opts, which in turn would set the scheduled state if opts contain :scheduled_at, effectively overriding the state value given in opts.

[sorentwo]

You're correct; the tests are generating an invalid query. There's not much of a reason to assert the scheduled time for something currently available, but I think we can still make it work.