feat: handles oban job exceptions

lib/tower.ex

@@ -11,12 +11,14 @@ defmodule Tower do
   def attach do
     :ok = Tower.LoggerHandler.attach()
     :ok = Tower.BanditExceptionHandler.attach()
+    :ok = Tower.ObanExceptionHandler.attach()
   end
 
   @spec detach() :: :ok
   def detach do
     :ok = Tower.LoggerHandler.detach()
     :ok = Tower.BanditExceptionHandler.detach()
+    :ok = Tower.ObanExceptionHandler.detach()
   end
 
   @spec handle_caught(Exception.kind(), Event.reason(), Exception.stacktrace()) :: :ok

lib/tower/oban_exception_handler.ex

@@ -0,0 +1,40 @@
+defmodule Tower.ObanExceptionHandler do
+  require Logger
+
+  @handler_id __MODULE__
+
+  def attach do
+    :telemetry.attach(
+      @handler_id,
+      [:oban, :job, :exception],
+      &__MODULE__.handle_event/4,
+      _handler_config = []
+    )
+  end
+
+  def detach do
+    :telemetry.detach(@handler_id)
+  end
+
+  def handle_event(
+        [:oban, :job, :exception],
+        _event_measurements,
+        %{kind: kind, reason: reason, stacktrace: stacktrace},
+        _handler_config
+      ) do
+    Tower.handle_caught(kind, reason, stacktrace)
+  end
+
+  def handle_event(
+        [:oban, :job, :exception],
+        _event_measurementes,
+        event_metadata,
+        _handler_config
+      ) do
+    Logger.warning(
+      "UNHANDLED OBAN JOB EXCEPTION with event_metadata=#{inspect(event_metadata, pretty: true)}"
+    )
+
+    :ignored
+  end
+end

mix.exs

@@ -50,7 +50,9 @@ defmodule Tower.MixProject do
       # Test
       {:assert_eventually, "~> 1.0", only: :test},
       {:plug_cowboy, "~> 2.7", only: :test},
-      {:bandit, "~> 1.5", only: :test}
+      {:bandit, "~> 1.5", only: :test},
+      {:oban, "~> 2.18", only: :test},
+      {:ecto_sqlite3, "~> 0.17.0", only: :test}
     ]
   end
 

mix.lock

@@ -1,19 +1,29 @@
 %{
   "assert_eventually": {:hex, :assert_eventually, "1.0.0", "f1539f28ba3ffa99a712433c77723c7103986932aa341d05eee94c333a920d15", [:mix], [{:ex_doc, ">= 0.0.0", [hex: :ex_doc, repo: "hexpm", optional: true]}], "hexpm", "c658ac4103c8bd82d0cf72a2fdb77477ba3fbc6b15228c5c801003d239625c69"},
   "bandit": {:hex, :bandit, "1.5.7", "6856b1e1df4f2b0cb3df1377eab7891bec2da6a7fd69dc78594ad3e152363a50", [:mix], [{:hpax, "~> 1.0.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "f2dd92ae87d2cbea2fa9aa1652db157b6cba6c405cb44d4f6dd87abba41371cd"},
+  "cc_precompiler": {:hex, :cc_precompiler, "0.1.10", "47c9c08d8869cf09b41da36538f62bc1abd3e19e41701c2cea2675b53c704258", [:mix], [{:elixir_make, "~> 0.7", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "f6e046254e53cd6b41c6bacd70ae728011aa82b2742a80d6e2214855c6e06b22"},
   "cowboy": {:hex, :cowboy, "2.12.0", "f276d521a1ff88b2b9b4c54d0e753da6c66dd7be6c9fca3d9418b561828a3731", [:make, :rebar3], [{:cowlib, "2.13.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "8a7abe6d183372ceb21caa2709bec928ab2b72e18a3911aa1771639bef82651e"},
   "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"},
   "cowlib": {:hex, :cowlib, "2.13.0", "db8f7505d8332d98ef50a3ef34b34c1afddec7506e4ee4dd4a3a266285d282ca", [:make, :rebar3], [], "hexpm", "e1e1284dc3fc030a64b1ad0d8382ae7e99da46c3246b815318a4b848873800a4"},
+  "db_connection": {:hex, :db_connection, "2.7.0", "b99faa9291bb09892c7da373bb82cba59aefa9b36300f6145c5f201c7adf48ec", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "dcf08f31b2701f857dfc787fbad78223d61a32204f217f15e881dd93e4bdd3ff"},
+  "decimal": {:hex, :decimal, "2.1.1", "5611dca5d4b2c3dd497dec8f68751f1f1a54755e8ed2a966c2633cf885973ad6", [:mix], [], "hexpm", "53cfe5f497ed0e7771ae1a475575603d77425099ba5faef9394932b35020ffcc"},
   "dialyxir": {:hex, :dialyxir, "1.4.3", "edd0124f358f0b9e95bfe53a9fcf806d615d8f838e2202a9f430d59566b6b53b", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "bf2cfb75cd5c5006bec30141b131663299c661a864ec7fbbc72dfa557487a986"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.40", "f3534689f6b58f48aa3a9ac850d4f05832654fe257bf0549c08cc290035f70d5", [:mix], [], "hexpm", "cdb34f35892a45325bad21735fadb88033bcb7c4c296a999bde769783f53e46a"},
+  "ecto": {:hex, :ecto, "3.12.1", "626765f7066589de6fa09e0876a253ff60c3d00870dd3a1cd696e2ba67bfceea", [:mix], [{:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "df0045ab9d87be947228e05a8d153f3e06e0d05ab10c3b3cc557d2f7243d1940"},
+  "ecto_sql": {:hex, :ecto_sql, "3.12.0", "73cea17edfa54bde76ee8561b30d29ea08f630959685006d9c6e7d1e59113b7d", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.12", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "dc9e4d206f274f3947e96142a8fdc5f69a2a6a9abb4649ef5c882323b6d512f0"},
+  "ecto_sqlite3": {:hex, :ecto_sqlite3, "0.17.0", "081c473dc662b42b8ce8f68c21fcf4a8dc52ce98ccc762cd6d45e583a7bdf445", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.11", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.11", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:exqlite, "~> 0.22", [hex: :exqlite, repo: "hexpm", optional: false]}], "hexpm", "68686c97accb1369012a957bf6da85d71db4b358529a4513695d78fce92b82f2"},
+  "elixir_make": {:hex, :elixir_make, "0.8.4", "4960a03ce79081dee8fe119d80ad372c4e7badb84c493cc75983f9d3bc8bde0f", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:certifi, "~> 2.0", [hex: :certifi, repo: "hexpm", optional: true]}], "hexpm", "6e7f1d619b5f61dfabd0a20aa268e575572b542ac31723293a4c1a567d5ef040"},
   "erlex": {:hex, :erlex, "0.2.7", "810e8725f96ab74d17aac676e748627a07bc87eb950d2b83acd29dc047a30595", [:mix], [], "hexpm", "3ed95f79d1a844c3f6bf0cea61e0d5612a42ce56da9c03f01df538685365efb0"},
   "ex_doc": {:hex, :ex_doc, "0.34.2", "13eedf3844ccdce25cfd837b99bea9ad92c4e511233199440488d217c92571e8", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "5ce5f16b41208a50106afed3de6a2ed34f4acfd65715b82a0b84b49d995f95c1"},
+  "exqlite": {:hex, :exqlite, "0.23.0", "6e851c937a033299d0784994c66da24845415072adbc455a337e20087bce9033", [:make, :mix], [{:cc_precompiler, "~> 0.1", [hex: :cc_precompiler, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.8", [hex: :elixir_make, repo: "hexpm", optional: false]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "404341cceec5e6466aaed160cf0b58be2019b60af82588c215e1224ebd3ec831"},
   "hpax": {:hex, :hpax, "1.0.0", "28dcf54509fe2152a3d040e4e3df5b265dcb6cb532029ecbacf4ce52caea3fd2", [:mix], [], "hexpm", "7f1314731d711e2ca5fdc7fd361296593fc2542570b3105595bb0bc6d0fad601"},
+  "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"},
   "makeup": {:hex, :makeup, "1.1.2", "9ba8837913bdf757787e71c1581c21f9d2455f4dd04cfca785c70bbfff1a76a3", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cce1566b81fbcbd21eca8ffe808f33b221f9eee2cbc7a1706fc3da9ff18e6cac"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.16.2", "627e84b8e8bf22e60a2579dad15067c755531fea049ae26ef1020cad58fe9578", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "41193978704763f6bbe6cc2758b84909e62984c7752b3784bd3c218bb341706b"},
   "makeup_erlang": {:hex, :makeup_erlang, "1.0.0", "6f0eff9c9c489f26b69b61440bf1b238d95badae49adac77973cbacae87e3c2e", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "ea7a9307de9d1548d2a72d299058d1fd2339e3d398560a0e46c27dab4891e4d2"},
   "mime": {:hex, :mime, "2.0.6", "8f18486773d9b15f95f4f4f1e39b710045fa1de891fada4516559967276e4dc2", [:mix], [], "hexpm", "c9945363a6b26d747389aac3643f8e0e09d30499a138ad64fe8fd1d13d9b153e"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"},
+  "oban": {:hex, :oban, "2.18.2", "583e78965ee15263ac968e38c983bad169ae55eadaa8e1e39912562badff93ba", [:mix], [{:ecto_sql, "~> 3.10", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "9dd25fd35883a91ed995e9fe516e479344d3a8623dfe2b8c3fc8e5be0228ec3a"},
   "plug": {:hex, :plug, "1.16.1", "40c74619c12f82736d2214557dedec2e9762029b2438d6d175c5074c933edc9d", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "a13ff6b9006b03d7e33874945b2755253841b238c34071ed85b0e86057f8cddc"},
   "plug_cowboy": {:hex, :plug_cowboy, "2.7.1", "87677ffe3b765bc96a89be7960f81703223fe2e21efa42c125fcd0127dd9d6b2", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "02dbd5f9ab571b864ae39418db7811618506256f6d13b4a45037e5fe78dc5de3"},
   "plug_crypto": {:hex, :plug_crypto, "2.1.0", "f44309c2b06d249c27c8d3f65cfe08158ade08418cf540fd4f72d4d6863abb7b", [:mix], [], "hexpm", "131216a4b030b8f8ce0f26038bc4421ae60e4bb95c5cf5395e1421437824c4fa"},

test/oban/tower_oban_test.exs

@@ -0,0 +1,105 @@
+defmodule TowerObanTest do
+  use ExUnit.Case
+  doctest Tower
+
+  use AssertEventually, timeout: 100, interval: 10
+
+  setup do
+    start_link_supervised!(Tower.EphemeralReporter)
+
+    start_link_supervised!({
+      TestApp.Repo,
+      database: "tmp/test-#{:rand.uniform(10_000)}.db", journal_mode: :memory
+    })
+
+    start_link_supervised!(
+      {Oban, engine: Oban.Engines.Lite, repo: TestApp.Repo, queues: [default: 10]}
+    )
+
+    Ecto.Migrator.up(TestApp.Repo, 0, TestApp.Repo.Migrations.AddOban)
+
+    Tower.attach()
+
+    on_exit(fn ->
+      Tower.detach()
+    end)
+  end
+
+  @tag capture_log: true
+  test "reports raised exception in an Oban worker" do
+    TestApp.ArithmeticErrorWorker.new(%{}, max_attempts: 1)
+    |> Oban.insert()
+
+    assert_eventually(
+      [
+        %{
+          id: id,
+          datetime: datetime,
+          level: :error,
+          kind: :error,
+          reason: %RuntimeError{message: "error from an Oban worker"},
+          stacktrace: stacktrace
+        }
+      ] = Tower.EphemeralReporter.events()
+    )
+
+    assert String.length(id) == 36
+    assert recent_datetime?(datetime)
+    assert is_list(stacktrace)
+  end
+
+  @tag capture_log: true
+  test "reports uncaught throw generated in an Oban worker" do
+    TestApp.UncaughtThrowWorker.new(%{}, max_attempts: 1)
+    |> Oban.insert()
+
+    assert_eventually(
+      [
+        %{
+          id: id,
+          datetime: datetime,
+          level: :error,
+          kind: :error,
+          reason: %Oban.CrashError{reason: "something"},
+          stacktrace: stacktrace
+        }
+      ] = Tower.EphemeralReporter.events()
+    )
+
+    assert String.length(id) == 36
+    assert recent_datetime?(datetime)
+    assert is_list(stacktrace)
+  end
+
+  @tag capture_log: true
+  test "reports abnormal exit generated in an Oban worker" do
+    TestApp.AbnormalExitWorker.new(%{}, max_attempts: 1)
+    |> Oban.insert()
+
+    assert_eventually(
+      [
+        %{
+          id: id,
+          datetime: datetime,
+          level: :error,
+          kind: :error,
+          reason: %Oban.CrashError{reason: :abnormal},
+          stacktrace: stacktrace
+        }
+      ] = Tower.EphemeralReporter.events()
+    )
+
+    assert String.length(id) == 36
+    assert recent_datetime?(datetime)
+    assert is_list(stacktrace)
+  end
+
+  defp recent_datetime?(datetime) do
+    diff =
+      :logger.timestamp()
+      |> DateTime.from_unix!(:microsecond)
+      |> DateTime.diff(datetime, :microsecond)
+
+    diff >= 0 && diff < 100_000
+  end
+end

test/support/test_app.ex

@@ -0,0 +1,44 @@
+defmodule TestApp.UncaughtThrowWorker do
+  use Oban.Worker
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    throw("something")
+
+    :ok
+  end
+end
+
+defmodule TestApp.AbnormalExitWorker do
+  use Oban.Worker
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    exit(:abnormal)
+
+    :ok
+  end
+end
+
+defmodule TestApp.ArithmeticErrorWorker do
+  use Oban.Worker
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    raise "error from an Oban worker"
+
+    :ok
+  end
+end
+
+defmodule TestApp.Repo.Migrations.AddOban do
+  use Ecto.Migration
+
+  def change do
+    Oban.Migrations.up()
+  end
+end
+
+defmodule TestApp.Repo do
+  use Ecto.Repo, otp_app: :test_app, adapter: Ecto.Adapters.SQLite3
+end

test/tower_test.exs

@@ -5,8 +5,8 @@
   use AssertEventually, timeout: 100, interval: 10
 
   setup do
-    Tower.attach()
     start_reporter()
+    Tower.attach()
 
     on_exit(fn ->
       Tower.detach()
@@ -20,7 +20,7 @@
   @tag capture_log: true
   test "reports arithmetic error" do
     in_unlinked_process(fn ->
       1 / 0
     end)
 
     assert_eventually(
@@ -289,7 +289,7 @@
   test "reports Exception manually" do
     in_unlinked_process(fn ->
       try do
         1 / 0
       catch
         kind, reason ->
           Tower.handle_caught(kind, reason, __STACKTRACE__)
@@ -315,7 +315,7 @@
   test "reports Exception manually (shorthand)" do
     in_unlinked_process(fn ->
       try do
         1 / 0
       rescue
         e ->
           Tower.handle_exception(e, __STACKTRACE__)