Repository: quantum-elixir/quantum-core (elixir)

Issue:
hidden duplicate job (or partial update) when calling add_job with named job (same name)
given this Scheduler

```
defmodule YourApp.Scheduler do
  use Quantum.Scheduler,
    otp_app: :your_app

  import Crontab.CronExpression

  def first_version() do
    new_job()
    |> Quantum.Job.set_name(:test)
    |> Quantum.Job.set_schedule(~e[*/5 * * * * *]e)
    |> Quantum.Job.set_task(fn ->
      IO.puts("#{Timex.now()} - first hello")
    end)
    |> add_job()
  end

  def second_version() do
    new_job()
    |> Quantum.Job.set_name(:test)
    |> Quantum.Job.set_schedule(~e[*/10 * * * * *]e)
    |> Quantum.Job.set_task(fn ->
      IO.puts("#{Timex.now()} - second hello")
    end)
    |> add_job()
  end

end
```

start by creating a job (iex prompts reorganized):

```
➤ iex -S mix
Erlang/OTP 21 [erts-10.1.1] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] [hipe]

Compiling 1 file (.ex)
Interactive Elixir (1.7.4) - press Ctrl+C to exit (type h() ENTER for help)
iex(1)> Logger.configure(level: :info)
:ok
iex(2)> import YourApp.Scheduler
YourApp.Scheduler
iex(3)> jobs()
[]
iex(4)> first_version()
:ok
2019-01-15 20:54:03.618109Z - first hello
2019-01-15 20:54:03.618139Z - first hello
2019-01-15 20:54:03.618127Z - first hello
2019-01-15 20:54:03.618139Z - first hello
2019-01-15 20:54:05.001217Z - first hello
2019-01-15 20:54:10.001246Z - first hello
iex(5)> jobs()
[
  test: %Quantum.Job{
    name: :test,
    overlap: true,
    run_strategy: %Quantum.RunStrategy.Random{nodes: :cluster},
    schedule: ~e[*/5 * * * * * *]e,
    state: :active,
    task: #Function<0.60421259/0 in YourApp.Scheduler.first_version/0>,
    timezone: :utc
  }
]
2019-01-15 20:54:15.001190Z - first hello
2019-01-15 20:54:20.001329Z - first hello
```

and then add a different job with the same name:

```
iex(6)> second_version()
:ok
2019-01-15 20:54:25.001364Z - first hello
2019-01-15 20:54:30.001232Z - first hello
2019-01-15 20:54:30.001357Z - second hello
2019-01-15 20:54:35.001256Z - first hello
2019-01-15 20:54:40.001469Z - first hello
2019-01-15 20:54:40.001632Z - second hello
2019-01-15 20:54:45.001235Z - first hello
iex(7)> jobs()
[
  test: %Quantum.Job{
    name: :test,
    overlap: true,
    run_strategy: %Quantum.RunStrategy.Random{nodes: :cluster},
    schedule: ~e[*/10 * * * * * *]e,
    state: :active,
    task: #Function<1.60421259/0 in YourApp.Scheduler.second_version/0>,
    timezone: :utc
  }
]
2019-01-15 20:54:50.001341Z - first hello
2019-01-15 20:54:50.001427Z - second hello
```

and then the second version again:

```
iex(8)> second_version()
:ok
2019-01-15 20:54:55.001055Z - first hello
2019-01-15 20:55:00.001276Z - first hello
2019-01-15 20:55:00.001342Z - second hello
2019-01-15 20:55:00.001369Z - second hello
iex(9)> jobs()
[
  test: %Quantum.Job{
    name: :test,
    overlap: true,
    run_strategy: %Quantum.RunStrategy.Random{nodes: :cluster},
    schedule: ~e[*/10 * * * * * *]e,
    state: :active,
    task: #Function<1.60421259/0 in YourApp.Scheduler.second_version/0>,
    timezone: :utc
  }
]
2019-01-15 20:55:05.001078Z - first hello
2019-01-15 20:55:10.001225Z - first hello
2019-01-15 20:55:10.001294Z - second hello
2019-01-15 20:55:10.001324Z - second hello
2019-01-15 20:55:15.001341Z - first hello
2019-01-15 20:55:20.001707Z - first hello
2019-01-15 20:55:20.001809Z - second hello
2019-01-15 20:55:20.001859Z - second hello
```

and finally deactivate and reactivate the job by name:

```
iex(10)> deactivate_job(:test)
:ok
iex(11)> activate_job(:test)
2019-01-15 20:55:33.518137Z - second hello
2019-01-15 20:55:40.001193Z - second hello
2019-01-15 20:55:50.001085Z - second hello
```

=> When adding a job with the same name as a previous job, both version will be running but only the second one with be visible when listing jobs with `jobs/0` (the same happens when adding the same job twice)

=> `deactivate_job/1` followed by `activate_job/1` fixes it, i.e. the zombies are gone

PS also when adding a job, it might be run multiple times before the first schedules, e.g.
```
2019-01-15 20:54:03.618109Z - first hello
2019-01-15 20:54:03.618139Z - first hello
2019-01-15 20:54:03.618127Z - first hello
2019-01-15 20:54:03.618139Z - first hello
```


Pull request description:
Solution: Properly override jobs with duplicate job name
fixes #392 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index dc4361c..67f57be 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,9 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 
+### Fixed
+- Properly override jobs with duplicate name (#392)
+
 Diff for [unreleased]
 
 ## 3.1.0 - 2020-08-18
diff --git a/lib/quantum/job_broadcaster.ex b/lib/quantum/job_broadcaster.ex
index 99f4d98..369d1e4 100644
--- a/lib/quantum/job_broadcaster.ex
+++ b/lib/quantum/job_broadcaster.ex
@@ -82,6 +82,7 @@ defmodule Quantum.JobBroadcaster do
   end
 
   @impl GenStage
+
   def handle_cast(
         {:add, %Job{state: :active, name: job_name} = job},
         %State{
@@ -91,14 +92,40 @@ defmodule Quantum.JobBroadcaster do
           debug_logging: debug_logging
         } = state
       ) do
-    debug_logging &&
-      Logger.debug(fn ->
-        "[#{inspect(Node.self())}][#{__MODULE__}] Adding job #{inspect(job_name)}"
-      end)
+    case jobs do
+      %{^job_name => %Job{state: :active} = old_job} ->
+        debug_logging &&
+          Logger.debug(fn ->
+            "[#{inspect(Node.self())}][#{__MODULE__}] Replacing job #{inspect(job_name)}"
+          end)
+
+        :ok = storage.delete_job(storage_pid, job_name)
+        :ok = storage.add_job(storage_pid, job)
+
+        {:noreply, [{:delete, old_job}, {:add, job}],
+         %{state | jobs: Map.put(jobs, job_name, job)}}
 
-    :ok = storage.add_job(storage_pid, job)
+      %{^job_name => %Job{state: :inactive}} ->
+        debug_logging &&
+          Logger.debug(fn ->
+            "[#{inspect(Node.self())}][#{__MODULE__}] Replacing job #{inspect(job_name)}"
+          end)
 
-    {:noreply, [{:add, job}], %{state | jobs: Map.put(jobs, job_name, job)}}
+        :ok = storage.delete_job(storage_pid, job_name)
+        :ok = storage.add_job(storage_pid, job)
+
+        {:noreply, [{:add, job}], %{state | jobs: Map.put(jobs, job_name, job)}}
+
+      _ ->
+        debug_logging &&
+          Logger.debug(fn ->
+            "[#{inspect(Node.self())}][#{__MODULE__}] Adding job #{inspect(job_name)}"
+          end)
+
+        :ok = storage.add_job(storage_pid, job)
+
+        {:noreply, [{:add, job}], %{state | jobs: Map.put(jobs, job_name, job)}}
+    end
   end
 
   def handle_cast(
@@ -110,14 +137,39 @@ defmodule Quantum.JobBroadcaster do
           debug_logging: debug_logging
         } = state
       ) do
-    debug_logging &&
-      Logger.debug(fn ->
-        "[#{inspect(Node.self())}][#{__MODULE__}] Adding job #{inspect(job_name)}"
-      end)
+    case jobs do
+      %{^job_name => %Job{state: :active} = old_job} ->
+        debug_logging &&
+          Logger.debug(fn ->
+            "[#{inspect(Node.self())}][#{__MODULE__}] Replacing job #{inspect(job_name)}"
+          end)
+
+        :ok = storage.delete_job(storage_pid, job_name)
+        :ok = storage.add_job(storage_pid, job)
+
+        {:noreply, [{:delete, old_job}], %{state | jobs: Map.put(jobs, job_name, job)}}
 
-    :ok = storage.add_job(storage_pid, job)
+      %{^job_name => %Job{state: :inactive}} ->
+        debug_logging &&
+          Logger.debug(fn ->
+            "[#{inspect(Node.self())}][#{__MODULE__}] Replacing job #{inspect(job_name)}"
+          end)
 
-    {:noreply, [], %{state | jobs: Map.put(jobs, job_name, job)}}
+        :ok = storage.delete_job(storage_pid, job_name)
+        :ok = storage.add_job(storage_pid, job)
+
+        {:noreply, [], %{state | jobs: Map.put(jobs, job_name, job)}}
+
+      _ ->
+        debug_logging &&
+          Logger.debug(fn ->
+            "[#{inspect(Node.self())}][#{__MODULE__}] Adding job #{inspect(job_name)}"
+          end)
+
+        :ok = storage.add_job(storage_pid, job)
+
+        {:noreply, [], %{state | jobs: Map.put(jobs, job_name, job)}}
+    end
   end
 
   def handle_cast(


Test patch:
diff --git a/test/quantum/job_broadcaster_test.exs b/test/quantum/job_broadcaster_test.exs
index c1c0abc..6a977e8 100644
--- a/test/quantum/job_broadcaster_test.exs
+++ b/test/quantum/job_broadcaster_test.exs
@@ -10,6 +10,8 @@ defmodule Quantum.JobBroadcasterTest do
   import ExUnit.CaptureLog
   import Quantum.CaptureLogExtend
 
+  import Crontab.CronExpression
+
   doctest JobBroadcaster
 
   defmodule TestScheduler do
@@ -163,6 +165,98 @@ defmodule Quantum.JobBroadcasterTest do
         assert_receive {:add_job, ^inactive_job, _}
       end)
     end
+
+    @tag listen_storage: true
+    test "override active with active", %{broadcaster: broadcaster, test: test_name} do
+      job_1 =
+        TestScheduler.new_job()
+        |> Quantum.Job.set_name(test_name)
+        |> Quantum.Job.set_schedule(~e[*/5 * * * * *]e)
+
+      TestScheduler.add_job(broadcaster, job_1)
+
+      assert_receive {:received, {:add, ^job_1}}
+
+      job_2 =
+        TestScheduler.new_job()
+        |> Quantum.Job.set_name(test_name)
+        |> Quantum.Job.set_schedule(~e[*/10 * * * * *]e)
+
+      TestScheduler.add_job(broadcaster, job_2)
+
+      assert_receive {:received, {:delete, ^job_1}}
+      assert_receive {:received, {:add, ^job_2}}
+    end
+
+    @tag listen_storage: true
+    test "override active with inactive", %{broadcaster: broadcaster, test: test_name} do
+      job_1 =
+        TestScheduler.new_job()
+        |> Quantum.Job.set_name(test_name)
+        |> Quantum.Job.set_schedule(~e[*/5 * * * * *]e)
+
+      TestScheduler.add_job(broadcaster, job_1)
+
+      assert_receive {:received, {:add, ^job_1}}
+
+      job_2 =
+        TestScheduler.new_job()
+        |> Quantum.Job.set_name(test_name)
+        |> Quantum.Job.set_schedule(~e[*/10 * * * * *]e)
+        |> Quantum.Job.set_state(:inactive)
+
+      TestScheduler.add_job(broadcaster, job_2)
+
+      assert_receive {:received, {:delete, ^job_1}}
+      refute_receive {:received, {:add, _}}
+    end
+
+    @tag listen_storage: true
+    test "override inactive with active", %{broadcaster: broadcaster, test: test_name} do
+      job_1 =
+        TestScheduler.new_job()
+        |> Quantum.Job.set_name(test_name)
+        |> Quantum.Job.set_schedule(~e[*/5 * * * * *]e)
+        |> Quantum.Job.set_state(:inactive)
+
+      TestScheduler.add_job(broadcaster, job_1)
+
+      refute_receive {:received, {:add, _}}
+
+      job_2 =
+        TestScheduler.new_job()
+        |> Quantum.Job.set_name(test_name)
+        |> Quantum.Job.set_schedule(~e[*/10 * * * * *]e)
+
+      TestScheduler.add_job(broadcaster, job_2)
+
+      refute_receive {:received, {:delete, _}}
+      assert_receive {:received, {:add, ^job_2}}
+    end
+
+    @tag listen_storage: true
+    test "override inactive with inactive", %{broadcaster: broadcaster, test: test_name} do
+      job_1 =
+        TestScheduler.new_job()
+        |> Quantum.Job.set_name(test_name)
+        |> Quantum.Job.set_schedule(~e[*/5 * * * * *]e)
+        |> Quantum.Job.set_state(:inactive)
+
+      TestScheduler.add_job(broadcaster, job_1)
+
+      refute_receive {:received, {:add, _}}
+
+      job_2 =
+        TestScheduler.new_job()
+        |> Quantum.Job.set_name(test_name)
+        |> Quantum.Job.set_schedule(~e[*/10 * * * * *]e)
+        |> Quantum.Job.set_state(:inactive)
+
+      TestScheduler.add_job(broadcaster, job_2)
+
+      refute_receive {:received, {:delete, _}}
+      refute_receive {:received, {:add, _}}
+    end
   end
 
   describe "delete" do