Verify migrations at startup in testing mode

When Oban starts in a testing mode, it now verifies that migrations have
been run and are up to date. This catches migration issues early in CI
rather than failing with confusing database errors during test execution
or worse, in production.

For Postgres, the check verifies the migration version is current, while
for SQLite and MySQL, the check verifies the `oban_jobs` table exists.

Addresses #1429

Author: sorentwo

lib/oban.ex

@@ -511,9 +511,17 @@ defmodule Oban do
   def start_link(opts) when is_list(opts) do
     conf = Config.new(opts)
 
+    verify_migrated!(conf)
+
     Supervisor.start_link(__MODULE__, conf, name: Registry.via(conf.name, nil, conf))
   end
 
+  defp verify_migrated!(conf) do
+    if conf.testing != :disabled do
+      Oban.Migration.verify_migrated!(repo: conf.repo, prefix: conf.prefix)
+    end
+  end
+
   @doc false
   @spec child_spec([option]) :: Supervisor.child_spec()
   def child_spec(opts) do

lib/oban/migration.ex

@@ -159,7 +159,18 @@ defmodule Oban.Migration do
       Oban.Migration.up(unlogged: false)
   """
   def up(opts \\ []) when is_list(opts) do
-    migrator().up(opts)
+    migrator(opts).up(opts)
+  end
+
+  @doc """
+  Get the current migration version.
+
+  ## Example
+
+      Oban.Migration.current_version()
+  """
+  def current_version(opts \\ []) when is_list(opts) do
+    migrator(opts).current_version()
   end
 
   @doc """
@@ -180,7 +191,7 @@ defmodule Oban.Migration do
       Oban.Migration.down(prefix: "payments")
   """
   def down(opts \\ []) when is_list(opts) do
-    migrator().down(opts)
+    migrator(opts).down(opts)
   end
 
   @doc """
@@ -191,11 +202,60 @@ defmodule Oban.Migration do
       Oban.Migration.migrated_version()
   """
   def migrated_version(opts \\ []) when is_list(opts) do
-    migrator().migrated_version(opts)
+    migrator(opts).migrated_version(opts)
+  end
+
+  @doc false
+  def verify_migrated!(opts \\ []) when is_list(opts) do
+    current = current_version(opts)
+    version = migrated_version(opts)
+
+    cond do
+      version == 0 ->
+        raise RuntimeError, """
+        Oban migrations have not been run. The oban_jobs table does not exist.
+
+        Run migrations before starting Oban:
+
+            mix ecto.migrate
+
+        Or add a migration:
+
+            defmodule MyApp.Repo.Migrations.AddOban do
+              use Ecto.Migration
+
+              def up, do: Oban.Migrations.up()
+              def down, do: Oban.Migrations.down()
+            end
+        """
+
+      version < current ->
+        raise RuntimeError, """
+        Oban migrations are outdated. Found version #{version}, but version #{current} is required.
+
+        Run migrations to update:
+
+            mix ecto.migrate
+
+        Or add a migration:
+
+            defmodule MyApp.Repo.Migrations.UpdateObanToV#{current} do
+              use Ecto.Migration
+
+              def up, do: Oban.Migrations.up(version: #{current})
+              def down, do: Oban.Migrations.down(version: #{version})
+            end
+        """
+
+      true ->
+        :ok
+    end
   end
 
-  defp migrator do
-    case repo().__adapter__() do
+  defp migrator(opts) do
+    repo = Keyword.get_lazy(opts, :repo, &repo/0)
+
+    case repo.__adapter__() do
       Ecto.Adapters.Postgres -> Oban.Migrations.Postgres
       Ecto.Adapters.SQLite3 -> Oban.Migrations.SQLite
       Ecto.Adapters.MyXQL -> Oban.Migrations.MyXQL

lib/oban/migrations/myxql.ex

@@ -5,6 +5,8 @@ defmodule Oban.Migrations.MyXQL do
 
   use Ecto.Migration
 
+  def current_version, do: 1
+
   @impl Oban.Migration
   def up(_opts) do
     states =
@@ -65,5 +67,21 @@ defmodule Oban.Migrations.MyXQL do
   end
 
   @impl Oban.Migration
-  def migrated_version(_opts), do: 0
+  def migrated_version(opts) do
+    repo = Keyword.get_lazy(opts, :repo, fn -> repo() end)
+    config = repo.config()
+    database = Keyword.fetch!(config, :database)
+
+    query = """
+    SELECT 1
+    FROM information_schema.TABLES
+    WHERE TABLE_SCHEMA = '#{database}'
+    AND TABLE_NAME = 'oban_jobs'
+    """
+
+    case repo.query(query, [], log: false) do
+      {:ok, %{rows: [[1]]}} -> 1
+      _ -> 0
+    end
+  end
 end

lib/oban/migrations/sqlite.ex

@@ -5,6 +5,8 @@ defmodule Oban.Migrations.SQLite do
 
   use Ecto.Migration
 
+  def current_version, do: 1
+
   @impl Oban.Migration
   def up(_opts) do
     create_if_not_exists table(:oban_jobs, primary_key: false) do
@@ -49,5 +51,16 @@ defmodule Oban.Migrations.SQLite do
   end
 
   @impl Oban.Migration
-  def migrated_version(_opts), do: 0
+  def migrated_version(opts) do
+    repo = Keyword.get_lazy(opts, :repo, fn -> repo() end)
+
+    query = """
+    SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'oban_jobs'
+    """
+
+    case repo.query(query, [], log: false) do
+      {:ok, %{rows: [[1]]}} -> 1
+      _ -> 0
+    end
+  end
 end

test/oban/migrations/myxql_test.exs

@@ -13,7 +13,7 @@ defmodule Oban.Migrations.MyXQLTest do
     end
   end
 
-  @moduletag :lite
+  @moduletag :dolphin
 
   defmodule Migration do
     use Ecto.Migration
@@ -27,9 +27,27 @@ defmodule Oban.Migrations.MyXQLTest do
     end
   end
 
-  test "migrating a mysql database" do
+  defp storage_up(_conf) do
     MigrationRepo.__adapter__().storage_up(MigrationRepo.config())
+  end
+
+  defp storage_down do
+    MigrationRepo.__adapter__().storage_down(MigrationRepo.config())
+  end
+
+  setup :storage_up
+
+  test "verifying that any migrations have ran" do
+    start_supervised!(MigrationRepo)
 
+    assert_raise RuntimeError, ~r/migrations have not been run/, fn ->
+      start_supervised_oban!(repo: MigrationRepo, testing: :manual)
+    end
+  after
+    storage_down()
+  end
+
+  test "migrating a mysql database" do
     start_supervised!(MigrationRepo)
 
     assert :ok = Ecto.Migrator.up(MigrationRepo, 1, Migration)
@@ -40,7 +58,7 @@ defmodule Oban.Migrations.MyXQLTest do
     refute table_exists?("oban_jobs")
     refute table_exists?("oban_peers")
   after
-    MigrationRepo.__adapter__().storage_down(MigrationRepo.config())
+    storage_down()
   end
 
   defp table_exists?(name) do

test/oban/migrations/postgres_test.exs

@@ -55,6 +55,24 @@ defmodule Oban.Migrations.PostgresTest do
 
   @base_version 20_300_000_000_000
 
+  test "verifying that any migrations have ran" do
+    assert_raise RuntimeError, ~r/migrations have not been run/, fn ->
+      start_supervised_oban!(repo: Repo, testing: :manual, prefix: "does_not_exist")
+    end
+  end
+
+  test "verifying the database is migrated to the required version" do
+    Application.put_env(:oban, :up_version, current_version() - 1)
+
+    assert :ok = Ecto.Migrator.up(UnboxedRepo, @base_version, StepMigration)
+
+    assert_raise RuntimeError, ~r/migrations are outdated/, fn ->
+      start_supervised_oban!(repo: UnboxedRepo, testing: :manual, prefix: "migrating")
+    end
+  after
+    clear_migrated()
+  end
+
   test "migrating up and down between specific versions" do
     for up <- initial_version()..current_version() do
       Application.put_env(:oban, :up_version, up)

test/oban/migrations/lite_test.exs test/oban/migrations/sqlite_test.exstest/oban/migrations/lite_test.exs renamed to test/oban/migrations/sqlite_test.exs

@@ -27,6 +27,14 @@ defmodule Oban.Migrations.SQLiteTest do
     end
   end
 
+  test "verifying that any migrations have ran" do
+    start_supervised!(MigrationRepo)
+
+    assert_raise RuntimeError, ~r/migrations have not been run/, fn ->
+      start_supervised_oban!(repo: MigrationRepo, testing: :manual)
+    end
+  end
+
   test "migrating a sqlite database" do
     start_supervised!(MigrationRepo)