Repository: quantum-elixir/quantum-core
Pull request #467 (2020-11-30 19:28:51)

Optional `update_job` callback in `Quantum.Storage`
Add an optional `update_job` callback in `Quantum.Storage`. If this is implemented then it will be called when a job gets updated, otherwise the previous behaviour (delete and then add the job) will be used.

Rationale: I have created a `postgres` storage and the jobs are linked to other tables through foreign keys. With the current implementation every time a job is updated it deletes the current job and adds it again. I want to be able to just update the properties that have changed and leave everything else untouched.

maennchen: Thanks a lot. ❤️ I'll merge when the CI passed.
coveralls: ## Pull Request Test Coverage Report for [Build 46a44c435babfa68866a30eae98989dc53cab0a4-PR-467](https://coveralls.io/builds/35335028)

* **10** of **10**   **(100.0%)**  changed or added relevant lines in **2** files are covered.
* **12** unchanged lines in **2** files lost coverage.
* Overall coverage decreased (**-3.6%**) to **83.287%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [lib/quantum/job.ex](https://coveralls.io/builds/35335028/source?filename=lib%2Fquantum%2Fjob.ex#L210) | 1 | 94.12% |
| [lib/quantum/execution_broadcaster.ex](https://coveralls.io/builds/35335028/source?filename=lib%2Fquantum%2Fexecution_broadcaster.ex#L102) | 11 | 63.79% |
<!-- | **Total:** | **12** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/35335028/badge)](https://coveralls.io/builds/35335028) |
| :-- | --: |
| Change from base [Build 02e519fe0710b7c64a34762e3d3034d3ffb61aef](https://coveralls.io/builds/34683366): |  -3.6% |
| Covered Lines: | 299 |
| Relevant Lines: | 359 |

---
##### 💛  - [Coveralls](https://coveralls.io)

pnezis: @maennchen Please have a look at the tests issue I opened. I had to disable the tests locally in order to properly test the changes. 
maennchen: @pnezis Do you have time to have a look by yourself and open a PR as well? I don't have time at the moment and it may take some time until I can look at it.

Patch:
diff --git a/lib/quantum/job_broadcaster.ex b/lib/quantum/job_broadcaster.ex
index 35ed6f9..608bd76 100644
--- a/lib/quantum/job_broadcaster.ex
+++ b/lib/quantum/job_broadcaster.ex
@@ -107,14 +107,7 @@ defmodule Quantum.JobBroadcaster do
             {"Replacing job", node: Node.self(), name: job_name}
           end)
 
-        # Send event to telemetry incase the end user wants to monitor events
-        :telemetry.execute([:quantum, :job, :update], %{}, %{
-          job: job,
-          scheduler: state.scheduler
-        })
-
-        :ok = storage.delete_job(storage_pid, job_name)
-        :ok = storage.add_job(storage_pid, job)
+        :ok = update_job(storage, storage_pid, job, state.scheduler)
 
         {:noreply, [{:delete, old_job}, {:add, job}],
          %{state | jobs: Map.put(jobs, job_name, job)}}
@@ -125,14 +118,7 @@ defmodule Quantum.JobBroadcaster do
             {"Replacing job", node: Node.self(), name: job_name}
           end)
 
-        # Send event to telemetry incase the end user wants to monitor events
-        :telemetry.execute([:quantum, :job, :update], %{}, %{
-          job: job,
-          scheduler: state.scheduler
-        })
-
-        :ok = storage.delete_job(storage_pid, job_name)
-        :ok = storage.add_job(storage_pid, job)
+        :ok = update_job(storage, storage_pid, job, state.scheduler)
 
         {:noreply, [{:add, job}], %{state | jobs: Map.put(jobs, job_name, job)}}
 
@@ -170,14 +156,7 @@ defmodule Quantum.JobBroadcaster do
             {"Replacing job", node: Node.self(), name: job_name}
           end)
 
-        # Send event to telemetry incase the end user wants to monitor events
-        :telemetry.execute([:quantum, :job, :update], %{}, %{
-          job: job,
-          scheduler: state.scheduler
-        })
-
-        :ok = storage.delete_job(storage_pid, job_name)
-        :ok = storage.add_job(storage_pid, job)
+        :ok = update_job(storage, storage_pid, job, state.scheduler)
 
         {:noreply, [{:delete, old_job}], %{state | jobs: Map.put(jobs, job_name, job)}}
 
@@ -187,14 +166,7 @@ defmodule Quantum.JobBroadcaster do
             {"Replacing job", node: Node.self(), name: job_name}
           end)
 
-        # Send event to telemetry incase the end user wants to monitor events
-        :telemetry.execute([:quantum, :job, :update], %{}, %{
-          job: job,
-          scheduler: state.scheduler
-        })
-
-        :ok = storage.delete_job(storage_pid, job_name)
-        :ok = storage.add_job(storage_pid, job)
+        :ok = update_job(storage, storage_pid, job, state.scheduler)
 
         {:noreply, [], %{state | jobs: Map.put(jobs, job_name, job)}}
 
@@ -361,4 +333,19 @@ defmodule Quantum.JobBroadcaster do
   def handle_info(_message, state) do
     {:noreply, [], state}
   end
+
+  defp update_job(storage, storage_pid, %Job{name: job_name} = job, scheduler) do
+    # Send event to telemetry incase the end user wants to monitor events
+    :telemetry.execute([:quantum, :job, :update], %{}, %{
+      job: job,
+      scheduler: scheduler
+    })
+
+    if function_exported?(storage, :update_job, 2) do
+      :ok = storage.update_job(storage_pid, job)
+    else
+      :ok = storage.delete_job(storage_pid, job_name)
+      :ok = storage.add_job(storage_pid, job)
+    end
+  end
 end
diff --git a/lib/quantum/storage.ex b/lib/quantum/storage.ex
index a55dac5..3604efc 100644
--- a/lib/quantum/storage.ex
+++ b/lib/quantum/storage.ex
@@ -76,4 +76,14 @@ defmodule Quantum.Storage do
   Purge all date from storage and go back to initial state.
   """
   @callback purge(storage_pid :: storage_pid) :: :ok
+
+  @doc """
+  Updates existing job in storage.
+
+  This callback is optional. If not implemented then the `c:delete_job/2`
+  and then the `c:add_job/2` callbacks will be called instead.
+  """
+  @callback update_job(storage_pid :: storage_pid, job :: Job.t()) :: :ok
+
+  @optional_callbacks update_job: 2
 end


Test patch:
diff --git a/test/quantum/job_broadcaster_test.exs b/test/quantum/job_broadcaster_test.exs
index 54a2bc3..37818dc 100644
--- a/test/quantum/job_broadcaster_test.exs
+++ b/test/quantum/job_broadcaster_test.exs
@@ -6,6 +6,7 @@ defmodule Quantum.JobBroadcasterTest do
 
   alias Quantum.{Job, JobBroadcaster, JobBroadcaster.StartOpts}
   alias Quantum.Storage.Test, as: TestStorage
+  alias Quantum.Storage.TestWithUpdate, as: TestStorageWithUpdate
   alias Quantum.TestConsumer
 
   import ExUnit.CaptureLog
@@ -87,6 +88,15 @@ defmodule Quantum.JobBroadcasterTest do
           []
       end
 
+    storage =
+      case tags[:storage] do
+        :with_update ->
+          TestStorageWithUpdate
+
+        _ ->
+          TestStorage
+      end
+
     broadcaster =
       if tags[:manual_dispatch] do
         nil
@@ -98,7 +108,7 @@ defmodule Quantum.JobBroadcasterTest do
                %StartOpts{
                  name: __MODULE__,
                  jobs: init_jobs,
-                 storage: TestStorage,
+                 storage: storage,
                  scheduler: TestScheduler,
                  debug_logging: true
                }}
@@ -262,7 +272,38 @@ defmodule Quantum.JobBroadcasterTest do
       TestScheduler.add_job(broadcaster, job_2)
 
       assert_receive {:received, {:delete, ^job_1}}
+      assert_receive {:delete_job, ^test_name, _}
+      assert_receive {:received, {:add, ^job_2}}
+      assert_receive {:add_job, ^job_2, _}
+    end
+
+    @tag listen_storage: true, storage: :with_update
+    test "storage with update does not add and delete job", %{
+      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}}
+      assert_receive {:add_job, ^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}}
+      refute_receive {:delete_job, ^test_name, _}
       assert_receive {:received, {:add, ^job_2}}
+      refute_receive {:add_job, ^job_2, _}
+      assert_receive {:update_job, ^job_2, _}
     end
 
     @tag listen_storage: true
diff --git a/test/support/test_storage.ex b/test/support/test_storage.ex
index a4c6934..6535991 100644
--- a/test/support/test_storage.ex
+++ b/test/support/test_storage.ex
@@ -114,3 +114,13 @@ defmodule Quantum.Storage.Test do
     end
   end
 end
+
+defmodule Quantum.Storage.TestWithUpdate do
+  @moduledoc """
+  Test implementation of a `Quantum.Storage` that overrides `c:update_job/2`.
+  """
+  use Quantum.Storage.Test
+
+  @impl Quantum.Storage
+  def update_job(_storage_pid, job), do: send_and_wait(:update_job, job)
+end