Check dolphin leadership after upsert query

MySQL always returns the number of entries attempted, even when nothing
was added. The previous match caused all nodes to believe they were the
leader. This uses a secondary query within the same transaction to
detect if the current instance is the leader.

Closes #1250

Author: sorentwo

lib/oban/peers/database.ex

@@ -102,7 +102,7 @@ defmodule Oban.Peers.Database do
         fun = fn ->
           state
           |> delete_expired_peers()
-          |> upsert_peer()
+          |> attempt_leadership()
         end
 
         case Repo.transaction(state.conf, fun, retry: 1) do
@@ -163,14 +163,7 @@ defmodule Oban.Peers.Database do
   end
 
   def handle_call(:get_leader, _from, %State{conf: conf} = state) do
-    query =
-      "oban_peers"
-      |> where([p], p.name == ^inspect(conf.name))
-      |> select([p], p.node)
-
-    leader = Repo.one(conf, query)
-
-    {:reply, leader, state}
+    {:reply, query_leader(conf), state}
   end
 
   # Helpers
@@ -199,11 +192,16 @@ defmodule Oban.Peers.Database do
     Repo.delete_all(conf, query)
   end
 
-  defp notify_down(%State{conf: conf}) do
-    Notifier.notify(conf, :leader, %{down: inspect(conf.name)})
+  defp query_leader(conf) do
+    query =
+      "oban_peers"
+      |> where([p], p.name == ^inspect(conf.name))
+      |> select([p], p.node)
+
+    Repo.one(conf, query)
   end
 
-  defp upsert_peer(%State{conf: conf} = state) do
+  defp attempt_leadership(%State{conf: conf} = state) do
     started_at = DateTime.utc_now()
     expires_at = DateTime.add(started_at, state.interval, :millisecond)
 
@@ -214,24 +212,46 @@ defmodule Oban.Peers.Database do
       expires_at: expires_at
     }
 
-    # MySQL only supports auto-inference and not conflict_target.
-    base_opts =
+    leader? =
       if state.conf.engine == Oban.Engines.Dolphin do
-        []
+        dolphin_insert(peer_data, state)
       else
-        [conflict_target: :name]
+        regular_upsert(peer_data, state)
       end
 
+    %{state | leader?: leader?}
+  end
+
+  defp regular_upsert(peer_data, state) do
     repo_opts =
       if state.leader? do
-        Keyword.put(base_opts, :on_conflict, set: [expires_at: expires_at])
+        [conflict_target: :name, on_conflict: [set: [expires_at: peer_data.expires_at]]]
       else
-        [on_conflict: :nothing]
+        [conflict_target: :name, on_conflict: :nothing]
       end
 
-    case Repo.insert_all(conf, "oban_peers", [peer_data], repo_opts) do
-      {0, nil} -> %{state | leader?: false}
-      {_, nil} -> %{state | leader?: true}
+    case Repo.insert_all(state.conf, "oban_peers", [peer_data], repo_opts) do
+      {0, nil} -> false
+      {_, nil} -> true
     end
   end
+
+  defp dolphin_insert(peer_data, state) do
+    repo_opts =
+      if state.leader? do
+        [on_conflict: [set: [expires_at: peer_data.expires_at]]]
+      else
+        [on_conflict: :nothing]
+      end
+
+    # MySQL always returns the number of entries attempted, even when nothing was added. We have
+    # to check the leader after update.
+    Repo.insert_all(state.conf, "oban_peers", [peer_data], repo_opts)
+
+    query_leader(state.conf) == peer_data.node
+  end
+
+  defp notify_down(%State{conf: conf}) do
+    Notifier.notify(conf, :leader, %{down: inspect(conf.name)})
+  end
 end

test/oban/peers/database_test.exs

@@ -6,23 +6,46 @@ defmodule Oban.Peers.DatabaseTest do
   alias Oban.Peer
   alias Oban.Peers.Database
   alias Oban.TelemetryHandler
+  alias Oban.Test.DolphinRepo
 
-  test "only a single peer is leader" do
-    TelemetryHandler.attach_events()
-
+  test "enforcing a single leader with the Basic engine" do
     name = start_supervised_oban!(peer: false)
-    conf = %{Oban.config(name) | peer: {Database, []}}
+    conf = Oban.config(name)
+
+    assert [_leader] =
+             [A, B, C]
+             |> Enum.map(&start_supervised_peer!(conf, &1))
+             |> Enum.filter(&Database.leader?/1)
+  end
+
+  @tag :dolphin
+  test "enforcing a single leader with the Dolphin engine" do
+    name = start_supervised_oban!(engine: Oban.Engines.Dolphin, peer: false, repo: DolphinRepo)
+    conf = Oban.config(name)
 
     assert [_leader] =
              [A, B, C]
-             |> Enum.map(&start_supervised!({Peer, conf: conf, name: &1}))
+             |> Enum.map(&start_supervised_peer!(conf, &1))
              |> Enum.filter(&Database.leader?/1)
+  end
+
+  defp start_supervised_peer!(conf, name) do
+    node = "web.#{name}"
+    conf = %{conf | node: node, peer: {Database, []}}
+
+    start_supervised!({Peer, conf: conf, name: name})
+  end
+
+  test "dispatching leadership election events" do
+    TelemetryHandler.attach_events()
+
+    start_supervised_oban!(peer: Database)
 
-    assert_received {:event, [:election, :start], _measure,
-                     %{leader: _, peer: Database, was_leader: nil}}
+    assert_receive {:event, [:election, :start], _measure,
+                    %{leader: false, peer: Database, was_leader: nil}}
 
-    assert_received {:event, [:election, :stop], _measure,
-                     %{leader: _, peer: Database, was_leader: false}}
+    assert_receive {:event, [:election, :stop], _measure,
+                    %{leader: true, peer: Database, was_leader: false}}
   end
 
   test "gracefully handling a missing oban_peers table" do