Read the Oban job id without a struct pattern so consumers type-check (2.0.4)

.github/workflows/ci.yml

@@ -168,8 +168,8 @@ jobs:
   # code-and-toolchain requirement, not a CI topology. No macos/windows
   # build legs; do not re-add them.
 
-  # Type-check gate on the public API (found the dead IP-SAN matcher on
-  # its first run — 2026-08-22).
+  # Type-check the public API and the consumer worker compiled from test/support.
+  # The PostgreSQL suite exercises that same generated worker at runtime.
   dialyzer:
     runs-on: ubuntu-latest
     env:

CHANGELOG.md

@@ -6,6 +6,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Unreleased
 
+## 2.0.4 — October 9, 2026
+
+### Fixed
+
+- Generated Oban workers no longer report Dialyzer pattern-match warnings for
+  unsaved uniqueness conflicts. Oban's job type declares a persisted ID but its
+  Basic engine can return an unsaved conflict; admission reads the ID through a
+  private map helper. Retry limits, not-persisted refusals, identity checks, and
+  runnable-state checks retain their behavior.
+- The type gate now compiles the same generated worker used by the PostgreSQL
+  suite. Real Oban.Basic transaction tests force an unsaved conflict, verify
+  retry after the winning transaction commits, and verify refusal after all
+  twenty retries.
+- Dependency currency: Ash 3.34.4 → 3.34.6 (resolves EEF-CVE-2026-101028 in the
+  locked graph) and AshPostgres 2.14.3 → 2.14.5.
+
 ## 2.0.3 — October 7, 2026
 
 ### Fixed

CONTRIBUTING.md

@@ -68,10 +68,14 @@ mix compile --warnings-as-errors
 mix credo --strict
 mix test
 ASH_HOOKS_POSTGRES=1 mix test --cover
-mix dialyzer
+MIX_ENV=test mix dialyzer
 ./scripts/check-currency.sh
 ```
 
+The type gate uses the test environment to compile the PostgreSQL suite's worker
+from `test/support/postgres_worker.ex`. This checks the consumer code injected by
+`use AshHooks.Worker` as well as the library modules.
+
 The coverage gate runs the complete suite with PostgreSQL and the real receiver
 available. It requires 100% in the coverage surface declared in `mix.exs`.
 Mix's exclusions apply to whole modules: generated DSL entities, named test fixtures,

docs/PUBLISHING.md

@@ -38,7 +38,7 @@ operator procedure.
 5. Execute each notebook against that candidate archive before publishing:
 
    ```sh
-   ./scripts/run-livebook.sh documentation/livebooks/get-started.livemd --archive _build/release/ash_hooks-2.0.1.tar
+   ./scripts/run-livebook.sh documentation/livebooks/get-started.livemd --archive _build/release/ash_hooks-2.0.4.tar
    ```
 
    The runner supplies the unpacked archive explicitly inside this checkout.

documentation/livebooks/get-started.livemd

@@ -16,7 +16,7 @@ receiver. The receiver must already answer at `http://127.0.0.1:52871`;
 set `LOCAL_WEBHOOK_TESTER_PORT` to use another existing receiver. See
 the [local receiver guide](https://github.com/baselabs/ash_hooks/blob/main/ops/local/README.md).
 The notebook starts no server.
-Uses ash_hooks **2.0.3**. Its examples exercise the library locally; vendor-origin
+Uses ash_hooks **2.0.4**. Its examples exercise the library locally; vendor-origin
 delivery and a production queue require your own application integration.
 
 <!-- livebook:{"break_marks":true} -->
@@ -26,7 +26,7 @@ delivery and a production queue require your own application integration.
 ```elixir
 package =
   case System.get_env("ASH_HOOKS_NOTEBOOK_SOURCE") do
-    nil -> {:ash_hooks, "== 2.0.3"}
+    nil -> {:ash_hooks, "== 2.0.4"}
     source -> {:ash_hooks, path: Path.expand(source)}
   end
 
@@ -50,7 +50,7 @@ Mix.install(
 # and matches how SQL data layers count.
 Application.put_env(:ash, :default_string_length_count, :codepoints)
 Application.put_env(:ash_hooks, :tour_inbound_secret, :crypto.strong_rand_bytes(32))
-"2.0.3" = Application.spec(:ash_hooks, :vsn) |> to_string()
+"2.0.4" = Application.spec(:ash_hooks, :vsn) |> to_string()
 ```
 
 ```elixir

lib/ash_hooks/worker.ex

@@ -315,30 +315,40 @@ defmodule AshHooks.Worker do
 
       defp insert_durably(changeset, expected, retries_left) do
         case Oban.insert(@ash_hooks_oban, changeset) do
-          {:ok, %Oban.Job{id: nil, conflict?: true}} when retries_left > 0 ->
-            Process.sleep(5)
-            insert_durably(changeset, expected, retries_left - 1)
-
           {:ok, %Oban.Job{} = job} ->
-            durable_admission(job, expected)
+            if job.conflict? == true and not job_persisted?(job) and retries_left > 0 do
+              Process.sleep(5)
+              insert_durably(changeset, expected, retries_left - 1)
+            else
+              durable_admission(job, expected)
+            end
 
           {:error, reason} ->
             {:error, reason}
         end
       end
 
-      defp durable_admission(%Oban.Job{id: id, state: state, args: args}, expected)
-           when not is_nil(id) and state in ["available", "scheduled", "executing", "retryable"] do
-        if Map.take(args, Map.keys(expected)) == expected do
-          :ok
-        else
-          {:error, :job_identity_mismatch}
+      # Oban.Job.t() omits the unsaved uniqueness-conflict case (id: nil).
+      # Read the map in a helper so consumer type checks retain that runtime case.
+      defp job_persisted?(job), do: not is_nil(Map.get(job, :id))
+
+      defp durable_admission(%Oban.Job{state: state, args: args} = job, expected) do
+        cond do
+          not job_persisted?(job) ->
+            {:error, :job_not_persisted}
+
+          state in ["available", "scheduled", "executing", "retryable"] ->
+            if Map.take(args, Map.keys(expected)) == expected do
+              :ok
+            else
+              {:error, :job_identity_mismatch}
+            end
+
+          true ->
+            {:error, :job_not_runnable}
         end
       end
 
-      defp durable_admission(%Oban.Job{id: nil}, _expected), do: {:error, :job_not_persisted}
-      defp durable_admission(_job, _expected), do: {:error, :job_not_runnable}
-
       # the tenant rides the args only when the ledger carries a
       # multitenancy attribute (undeclared → nil → no key)
       # the row holds the ATTRIBUTE value; the args carry the TENANT —

mix.exs

@@ -1,7 +1,7 @@
 defmodule AshHooks.MixProject do
   use Mix.Project
 
-  @version "2.0.3"
+  @version "2.0.4"
   @source_url "https://github.com/baselabs/ash_hooks"
 
   def project do

mix.lock

@@ -1,6 +1,6 @@
 %{
-  "ash": {:hex, :ash, "3.34.4", "b799d852dca874b4afbe3bfa3477099e23b58830250fa6144c48ef7f58cf67d0", [:mix], [{:crux, ">= 0.1.2 and < 1.0.0-0", [hex: :crux, repo: "hexpm", optional: false]}, {:decimal, "~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.14", [hex: :ecto, repo: "hexpm", optional: false]}, {:ets, "~> 0.8", [hex: :ets, repo: "hexpm", optional: false]}, {:igniter, ">= 0.6.29 and < 1.0.0-0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, ">= 1.0.0", [hex: :jason, repo: "hexpm", optional: false]}, {:picosat_elixir, "~> 0.2", [hex: :picosat_elixir, repo: "hexpm", optional: true]}, {:plug, ">= 0.0.0", [hex: :plug, repo: "hexpm", optional: true]}, {:reactor, "~> 1.0", [hex: :reactor, repo: "hexpm", optional: false]}, {:simple_sat, ">= 0.1.1 and < 1.0.0-0", [hex: :simple_sat, repo: "hexpm", optional: true]}, {:spark, ">= 2.7.6 and < 3.0.0-0", [hex: :spark, repo: "hexpm", optional: false]}, {:splode, "~> 0.3", [hex: :splode, repo: "hexpm", optional: false]}, {:stream_data, "~> 1.0", [hex: :stream_data, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.1", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "bb82f7e412fce03b50abbe87a260ce956d1f67bc97d0544f011fa0d1882f07e4"},
-  "ash_postgres": {:hex, :ash_postgres, "2.14.3", "729e304de14bd5928f9df9545f3a29b849ba0a4b5fd65a8cf6f9416cfbad21fa", [:mix], [{:ash, "~> 3.34", [hex: :ash, repo: "hexpm", optional: false]}, {:ash_sql, "~> 0.8", [hex: :ash_sql, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.13", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:igniter, ">= 0.6.29 and < 1.0.0-0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, ">= 0.0.0", [hex: :postgrex, repo: "hexpm", optional: false]}, {:spark, ">= 2.3.4 and < 3.0.0-0", [hex: :spark, repo: "hexpm", optional: false]}], "hexpm", "afd5cad3d39aa56e31022e37c311522ac82bf25664ff20de141fda8ad4c4f362"},
+  "ash": {:hex, :ash, "3.34.6", "70ce402edbe6930570ce930661859b3be514dfd34700f8432c565e526ba9a990", [:mix], [{:crux, ">= 0.1.2 and < 1.0.0-0", [hex: :crux, repo: "hexpm", optional: false]}, {:decimal, "~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.14", [hex: :ecto, repo: "hexpm", optional: false]}, {:ets, "~> 0.8", [hex: :ets, repo: "hexpm", optional: false]}, {:igniter, ">= 0.6.29 and < 1.0.0-0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, ">= 1.0.0", [hex: :jason, repo: "hexpm", optional: false]}, {:picosat_elixir, "~> 0.2", [hex: :picosat_elixir, repo: "hexpm", optional: true]}, {:plug, ">= 0.0.0", [hex: :plug, repo: "hexpm", optional: true]}, {:reactor, "~> 1.0", [hex: :reactor, repo: "hexpm", optional: false]}, {:simple_sat, ">= 0.1.1 and < 1.0.0-0", [hex: :simple_sat, repo: "hexpm", optional: true]}, {:spark, ">= 2.7.6 and < 3.0.0-0", [hex: :spark, repo: "hexpm", optional: false]}, {:splode, "~> 0.3", [hex: :splode, repo: "hexpm", optional: false]}, {:stream_data, "~> 1.0", [hex: :stream_data, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.1", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "1109afed2bcf1f405aa035013d016a1ca88dab485ba2fb0c0885e7e7e9c6ae0f"},
+  "ash_postgres": {:hex, :ash_postgres, "2.14.5", "20a66d3d567817abc31117d0f5fc4a9241d835aa6504763475aa856ab12aadd4", [:mix], [{:ash, ">= 3.34.5 and < 4.0.0-0", [hex: :ash, repo: "hexpm", optional: false]}, {:ash_sql, "~> 0.8", [hex: :ash_sql, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.13", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:igniter, ">= 0.6.29 and < 1.0.0-0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, ">= 0.0.0", [hex: :postgrex, repo: "hexpm", optional: false]}, {:spark, ">= 2.3.4 and < 3.0.0-0", [hex: :spark, repo: "hexpm", optional: false]}], "hexpm", "e2993a5ba796b613870e7ee5d47f7e7a007e87061d475a916d1158b58fcd8fdd"},
   "ash_sql": {:hex, :ash_sql, "0.8.1", "ac473ebf772701b96a2378629d014af0ff615bffc16327d899a88dc0163ed278", [:mix], [{:ash, "~> 3.34", [hex: :ash, repo: "hexpm", optional: false]}, {:ecto, ">= 3.13.4 and < 4.0.0-0", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.9", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "f35fce068eac736278f8368119e7728a933e0be75c82bc1037f09ce5ad3c0b4b"},
   "ash_sqlite": {:hex, :ash_sqlite, "0.2.19", "8cacd83f618030ca97d22ef4d0ef2a4cfbaad4fa770144392706895b87f1051f", [:mix], [{:ash, "~> 3.33", [hex: :ash, repo: "hexpm", optional: false]}, {:ash_sql, ">= 0.6.9 and < 1.0.0-0", [hex: :ash_sql, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_libsql, "~> 0.9", [hex: :ecto_libsql, repo: "hexpm", optional: true]}, {:ecto_sql, "~> 3.13", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.12", [hex: :ecto_sqlite3, repo: "hexpm", optional: false]}, {:igniter, ">= 0.6.14 and < 1.0.0-0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "a25a8425d31a9bfb0784cf295cbd3ce5e33ebfe3b97622a50e279f6820638687"},
   "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},

scripts/check-ash-floor.exs

@@ -96,7 +96,9 @@ try do
     )
 
     Mix.Task.run("loadconfig")
-    Mix.Task.run("deps.unlock", ["ash"])
+    # Resolve the PostgreSQL test substrate for the exact Ash floor as well.
+    # The latest locked AshPostgres requires a newer Ash; the root lock stays current.
+    Mix.Task.run("deps.unlock", ["ash", "ash_postgres"])
     Mix.Task.run("deps.get")
     Mix.Task.run("compile", ["--warnings-as-errors"])
     Mix.Task.run("app.start")

test/ash_hooks/outbound_postgres_readiness_test.exs

@@ -142,16 +142,6 @@ if Code.ensure_loaded?(Oban) do
       def local_receiver?(url), do: URI.parse(url).host == "127.0.0.1"
     end
 
-    defmodule Worker do
-      @moduledoc false
-      use AshHooks.Worker,
-        deliveries: AshHooks.OutboundPostgresReadinessTest.Delivery,
-        endpoints: AshHooks.OutboundPostgresReadinessTest.Endpoint,
-        secret_resolver: {AshHooks.OutboundPostgresReadinessTest.Runtime, :secret},
-        oban: AshHooks.OutboundPostgresReadinessTest.Oban,
-        queue: :outbound_readiness
-    end
-
     defmodule OtherWorker do
       @moduledoc false
       use AshHooks.Worker,
@@ -257,6 +247,7 @@ if Code.ensure_loaded?(Oban) do
     alias AshHooks.Delivery, as: DeliveryRuntime
     alias AshHooks.{Dispatcher, Event, OutboundBinding, PrimaryKey}
     alias AshHooks.Http.Bounded
+    alias AshHooks.OutboundPostgresReadinessTest.Worker
     alias AshHooks.TestPostgres.Repo
 
     @moduletag :postgres
@@ -742,7 +733,9 @@ if Code.ensure_loaded?(Oban) do
         assert {:snooze, _seconds} =
                  DeliveryRuntime.run(
                    delivery_args(row),
-                   delivery_config(attempt_timeout: 100, finalization_allowance: 1_000)
+                   # Leave time for real PostgreSQL admission and endpoint reads.
+                   # The receiver's two-second delay still exceeds this budget.
+                   delivery_config(attempt_timeout: 1_000, finalization_allowance: 1_000)
                  )
 
         assert_receive_count(session, event.id, 1)
@@ -855,6 +848,46 @@ if Code.ensure_loaded?(Oban) do
         end
       end
 
+      @tag :unsaved_conflict
+      test "an unsaved conflict retries until the real winning transaction commits" do
+        {_endpoint, _session} = endpoint_and_subscription!()
+        event = event!()
+        row = dispatched_row!(event)
+        route = OutboundBinding.named_route(Worker, :enqueue)
+        assert {:ok, bound} = AshHooks.Worker.bind_route(row, route, nil)
+
+        with_uncommitted_job(bound, event, fn holder ->
+          observe_admission_jobs(holder)
+          assert :ok = Worker.enqueue(bound, event)
+          assert_receive {:admission_job, nil, true}
+          assert_receive {:admission_job, job_id, true} when is_integer(job_id)
+        end)
+
+        assert [[job_id, "available", _args]] = jobs_for_event(event.id)
+        assert is_integer(job_id)
+      end
+
+      @tag :unsaved_conflict
+      test "an unsaved conflict refuses admission after all twenty retries" do
+        {_endpoint, _session} = endpoint_and_subscription!()
+        event = event!()
+        row = dispatched_row!(event)
+        route = OutboundBinding.named_route(Worker, :enqueue)
+        assert {:ok, bound} = AshHooks.Worker.bind_route(row, route, nil)
+
+        with_uncommitted_job(bound, event, fn _holder ->
+          observe_admission_jobs(nil)
+          assert {:error, :job_not_persisted} = Worker.enqueue(bound, event)
+
+          for _ <- 1..21, do: assert_receive({:admission_job, nil, true})
+          refute_receive {:admission_job, _id, _conflict}, 0
+          assert [] = jobs_for_event(event.id)
+        end)
+
+        assert [[job_id, "available", _args]] = jobs_for_event(event.id)
+        assert is_integer(job_id)
+      end
+
       test "an advisory-lock uniqueness conflict resolves to one persisted job id" do
         {_endpoint, _session} = endpoint_and_subscription!()
         event = event!()
@@ -1131,6 +1164,61 @@ if Code.ensure_loaded?(Oban) do
       Ash.get!(Endpoint, endpoint.id, authorize?: false)
     end
 
+    # Hold Oban.Basic's actual uniqueness lock and inserted row in an open
+    # PostgreSQL transaction. A second connection must get Oban's unsaved job.
+    defp with_uncommitted_job(row, event, fun) do
+      parent = self()
+
+      holder =
+        Task.async(fn ->
+          Repo.transaction(fn ->
+            assert :ok = Worker.enqueue(row, event)
+            send(parent, :unique_job_uncommitted)
+
+            receive do
+              :commit -> :ok
+            after
+              5_000 -> raise "uniqueness transaction was not released"
+            end
+          end)
+        end)
+
+      try do
+        assert_receive :unique_job_uncommitted, 2_000
+        fun.(holder.pid)
+      after
+        send(holder.pid, :commit)
+        assert {:ok, :ok} = Task.await(holder, 5_000)
+      end
+    end
+
+    defp observe_admission_jobs(commit_on_conflict) do
+      handler = {__MODULE__, :admission_jobs, make_ref()}
+      parent = self()
+
+      :ok =
+        :telemetry.attach(
+          handler,
+          [:oban, :engine, :insert_job, :stop],
+          &__MODULE__.admission_job/4,
+          {parent, commit_on_conflict}
+        )
+
+      on_exit(fn -> :telemetry.detach(handler) end)
+    end
+
+    @doc false
+    def admission_job(_event, _measurements, %{job: job}, {parent, commit_on_conflict}) do
+      if self() == parent do
+        id = Map.get(job, :id)
+        send(parent, {:admission_job, id, job.conflict?})
+
+        if is_nil(id) and commit_on_conflict do
+          send(commit_on_conflict, :commit)
+        end
+      end
+    end
+
     defp event! do
       {:ok, event} = Event.new(type: :order_paid, payload: @payload)
       event
@@ -1189,8 +1277,10 @@ if Code.ensure_loaded?(Oban) do
           secret_resolver: {Runtime, :secret},
           http_opts: [validate_destination: false, timeout: 5_000],
           ssrf_check: fn url -> URI.parse(url).host == "127.0.0.1" end,
-          attempt_timeout: 3_000,
-          finalization_allowance: 1_000,
+          # Use the driver's normal budgets for real database and receiver work.
+          # Deadline tests override them explicitly.
+          attempt_timeout: 25_000,
+          finalization_allowance: 5_000,
           max_attempts: 3,
           base_backoff_seconds: 1,
           max_backoff_seconds: 2,

test/consumer/verify.exs

@@ -51,7 +51,7 @@ unless asymmetric_id == event.id, do: raise("asymmetric signature identity misma
   AshHooks.Http.Bounded.request(
     :get,
     "https://hex.pm/api/packages/ash_hooks",
-    %{"user-agent" => "ash_hooks-consumer-verification/2.0.2"},
+    %{"user-agent" => "ash_hooks-consumer-verification/#{version}"},
     "",
     max_body_bytes: 1_024,
     timeout: 10_000
@@ -63,7 +63,7 @@ unless byte_size(body) > 0 and byte_size(body) <= 1_024, do: raise("unexpected H
   AshHooks.Http.Httpc.request(
     :get,
     "https://hex.pm/api/packages/ash_hooks",
-    %{"user-agent" => "ash_hooks-consumer-verification/2.0.2"},
+    %{"user-agent" => "ash_hooks-consumer-verification/#{version}"},
     "",
     max_body_bytes: 1_024,
     timeout: 10_000

test/support/postgres_worker.ex

@@ -0,0 +1,13 @@
+# Compile the same worker used by the PostgreSQL suite so the type gate
+# analyzes the consumer code injected by AshHooks.Worker.
+if Code.ensure_loaded?(Oban) do
+  defmodule AshHooks.OutboundPostgresReadinessTest.Worker do
+    @moduledoc false
+    use AshHooks.Worker,
+      deliveries: AshHooks.OutboundPostgresReadinessTest.Delivery,
+      endpoints: AshHooks.OutboundPostgresReadinessTest.Endpoint,
+      secret_resolver: {AshHooks.OutboundPostgresReadinessTest.Runtime, :secret},
+      oban: AshHooks.OutboundPostgresReadinessTest.Oban,
+      queue: :outbound_readiness
+  end
+end