Repository: quantum-elixir/quantum-core
Pull request #429 (2020-02-28 19:18:18)

Process Based Storage Adapter


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 991d908..cf35040 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,9 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 
+### Changed
+- The `Quantum.Storage` behaviour contains a new  mandatory `child_spec/1` callback.
+
 Diff for [unreleased]
 
 ## 3.0.0-rc.1 - 2020-02-26
diff --git a/lib/quantum.ex b/lib/quantum.ex
index f110595..5b6eca8 100644
--- a/lib/quantum.ex
+++ b/lib/quantum.ex
@@ -52,7 +52,8 @@ defmodule Quantum do
     state: :active,
     timezone: :utc,
     run_strategy: {Random, :cluster},
-    debug_logging: true
+    debug_logging: true,
+    storage: Noop
   ]
 
   @optional_callbacks init: 1
@@ -135,18 +136,12 @@ defmodule Quantum do
 
   @doc false
   # Retrieves only scheduler related configuration.
-  def scheduler_config(quantum, otp_app, custom) do
-    config =
-      @defaults
-      |> Keyword.merge(Application.get_env(otp_app, quantum, []))
-      |> Keyword.merge(custom)
-      |> Keyword.merge(otp_app: otp_app, quantum: quantum)
-
-    config
-    |> Keyword.put_new(:quantum, quantum)
-    |> Keyword.put_new(:scheduler, quantum)
+  def scheduler_config(scheduler, otp_app, custom) do
+    @defaults
+    |> Keyword.merge(Application.get_env(otp_app, scheduler, []))
+    |> Keyword.merge(custom)
+    |> Keyword.merge(otp_app: otp_app, scheduler: scheduler)
     |> update_in([:schedule], &Normalizer.normalize_schedule/1)
-    |> Keyword.put_new(:storage, Noop)
   end
 
   defmacro __using__(opts) do
@@ -166,7 +161,7 @@ defmodule Quantum do
 
       defp __job_broadcaster__ do
         __job_broadcaster__(
-          config() |> Keyword.fetch!(:quantum) |> Module.concat(JobBroadcaster),
+          config() |> Keyword.fetch!(:scheduler) |> Module.concat(JobBroadcaster),
           config()
         )
       end
@@ -178,7 +173,8 @@ defmodule Quantum do
       defp __timeout__, do: Keyword.fetch!(config(), :timeout)
 
       @impl behaviour
-      def start_link(opts \\ [name: __MODULE__]) do
+      def start_link(opts \\ []) do
+        opts = Keyword.put_new(opts, :name, __MODULE__)
         Quantum.Supervisor.start_link(__MODULE__, @otp_app, opts)
       end
 
diff --git a/lib/quantum/clock_broadcaster.ex b/lib/quantum/clock_broadcaster.ex
index 5d4fb17..942a5cd 100644
--- a/lib/quantum/clock_broadcaster.ex
+++ b/lib/quantum/clock_broadcaster.ex
@@ -13,14 +13,29 @@ defmodule Quantum.ClockBroadcaster do
   def start_link(%StartOpts{name: name} = opts) do
     GenStage.start_link(
       __MODULE__,
-      struct!(InitOpts, Map.take(opts, [:start_time, :debug_logging])),
+      struct!(InitOpts, Map.take(opts, [:start_time, :storage, :scheduler, :debug_logging])),
       name: name
     )
   end
 
   @impl GenStage
   @spec init(opts :: InitOpts.t()) :: {:producer, State.t()}
-  def init(%InitOpts{debug_logging: debug_logging, start_time: start_time}) do
+  def init(%InitOpts{
+        debug_logging: debug_logging,
+        storage: storage,
+        scheduler: scheduler,
+        start_time: start_time
+      }) do
+    start_time =
+      scheduler
+      |> Module.concat(Storage)
+      |> GenServer.whereis()
+      |> storage.last_execution_date()
+      |> case do
+        :unknown -> start_time
+        date -> date
+      end
+
     {:producer,
      %State{
        time: %{start_time | microsecond: {0, 0}},
diff --git a/lib/quantum/clock_broadcaster/init_opts.ex b/lib/quantum/clock_broadcaster/init_opts.ex
index 4abf759..66629de 100644
--- a/lib/quantum/clock_broadcaster/init_opts.ex
+++ b/lib/quantum/clock_broadcaster/init_opts.ex
@@ -3,12 +3,16 @@ defmodule Quantum.ClockBroadcaster.InitOpts do
 
   # Init Options
 
+  alias Quantum.{Scheduler, Storage}
+
   @type t :: %__MODULE__{
           start_time: NaiveDateTime.t(),
+          storage: Storage,
+          scheduler: Scheduler,
           debug_logging: boolean()
         }
 
-  @enforce_keys [:start_time, :debug_logging]
+  @enforce_keys [:start_time, :storage, :scheduler, :debug_logging]
 
   defstruct @enforce_keys
 end
diff --git a/lib/quantum/clock_broadcaster/start_opts.ex b/lib/quantum/clock_broadcaster/start_opts.ex
index ec867f0..b6150a0 100644
--- a/lib/quantum/clock_broadcaster/start_opts.ex
+++ b/lib/quantum/clock_broadcaster/start_opts.ex
@@ -3,13 +3,17 @@ defmodule Quantum.ClockBroadcaster.StartOpts do
 
   # Start Options
 
+  alias Quantum.{Scheduler, Storage}
+
   @type t :: %__MODULE__{
           name: GenServer.server(),
           start_time: NaiveDateTime.t(),
+          storage: Storage,
+          scheduler: Scheduler,
           debug_logging: boolean()
         }
 
-  @enforce_keys [:name, :start_time, :debug_logging]
+  @enforce_keys [:start_time, :name, :storage, :scheduler, :debug_logging]
 
   defstruct @enforce_keys
 end
diff --git a/lib/quantum/execution_broadcaster.ex b/lib/quantum/execution_broadcaster.ex
index 9819a53..c6a6921 100644
--- a/lib/quantum/execution_broadcaster.ex
+++ b/lib/quantum/execution_broadcaster.ex
@@ -60,11 +60,14 @@ defmodule Quantum.ExecutionBroadcaster do
         scheduler: scheduler,
         debug_logging: debug_logging
       }) do
+    storage_pid = GenServer.whereis(Module.concat(scheduler, Storage))
+
     {:producer_consumer,
      %State{
        uninitialized_jobs: [],
        execution_timeline: [],
        storage: storage,
+       storage_pid: storage_pid,
        scheduler: scheduler,
        debug_logging: debug_logging
      }, subscribe_to: [job_broadcaster, clock_broadcaster]}
@@ -159,7 +162,7 @@ defmodule Quantum.ExecutionBroadcaster do
   defp execute_events_to_fire(
          %State{
            storage: storage,
-           scheduler: scheduler,
+           storage_pid: storage_pid,
            debug_logging: debug_logging,
            execution_timeline: [{time_to_execute, jobs} | tail]
          } = state,
@@ -173,7 +176,7 @@ defmodule Quantum.ExecutionBroadcaster do
         {[], state}
 
       :eq ->
-        :ok = storage.update_last_execution_date(scheduler, time_to_execute)
+        :ok = storage.update_last_execution_date(storage_pid, time_to_execute)
 
         events =
           for %Job{name: job_name} = job <- jobs do
diff --git a/lib/quantum/execution_broadcaster/state.ex b/lib/quantum/execution_broadcaster/state.ex
index 5113542..20af38c 100644
--- a/lib/quantum/execution_broadcaster/state.ex
+++ b/lib/quantum/execution_broadcaster/state.ex
@@ -10,6 +10,7 @@ defmodule Quantum.ExecutionBroadcaster.State do
           uninitialized_jobs: [Job.t()],
           execution_timeline: [{NaiveDateTime.t(), [Job.t()]}],
           storage: StorageAdapter,
+          storage_pid: StorageAdapter.storage_pid(),
           scheduler: Quantum,
           debug_logging: boolean()
         }
@@ -18,6 +19,7 @@ defmodule Quantum.ExecutionBroadcaster.State do
     :uninitialized_jobs,
     :execution_timeline,
     :storage,
+    :storage_pid,
     :scheduler,
     :debug_logging
   ]
diff --git a/lib/quantum/job_broadcaster.ex b/lib/quantum/job_broadcaster.ex
index 8147807..5635f49 100644
--- a/lib/quantum/job_broadcaster.ex
+++ b/lib/quantum/job_broadcaster.ex
@@ -29,8 +29,10 @@ defmodule Quantum.JobBroadcaster do
         scheduler: scheduler,
         debug_logging: debug_logging
       }) do
+    storage_pid = GenServer.whereis(Module.concat(scheduler, Storage))
+
     effective_jobs =
-      scheduler
+      storage_pid
       |> storage.jobs()
       |> case do
         :not_applicable ->
@@ -55,6 +57,7 @@ defmodule Quantum.JobBroadcaster do
        jobs: effective_jobs |> Enum.map(&{&1.name, &1}) |> Map.new(),
        buffer: for(%{state: :active} = job <- effective_jobs, do: {:add, job}),
        storage: storage,
+       storage_pid: storage_pid,
        scheduler: scheduler,
        debug_logging: debug_logging
      }}
@@ -70,38 +73,50 @@ defmodule Quantum.JobBroadcaster do
   @impl GenStage
   def handle_cast(
         {:add, %Job{state: :active, name: job_name} = job},
-        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
-          state
+        %State{
+          jobs: jobs,
+          storage: storage,
+          storage_pid: storage_pid,
+          debug_logging: debug_logging
+        } = state
       ) do
     debug_logging &&
       Logger.debug(fn ->
         "[#{inspect(Node.self())}][#{__MODULE__}] Adding job #{inspect(job_name)}"
       end)
 
-    :ok = storage.add_job(scheduler, job)
+    :ok = storage.add_job(storage_pid, job)
 
     {:noreply, [{:add, job}], %{state | jobs: Map.put(jobs, job_name, job)}}
   end
 
   def handle_cast(
         {:add, %Job{state: :inactive, name: job_name} = job},
-        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
-          state
+        %State{
+          jobs: jobs,
+          storage: storage,
+          storage_pid: storage_pid,
+          debug_logging: debug_logging
+        } = state
       ) do
     debug_logging &&
       Logger.debug(fn ->
         "[#{inspect(Node.self())}][#{__MODULE__}] Adding job #{inspect(job_name)}"
       end)
 
-    :ok = storage.add_job(scheduler, job)
+    :ok = storage.add_job(storage_pid, job)
 
     {:noreply, [], %{state | jobs: Map.put(jobs, job_name, job)}}
   end
 
   def handle_cast(
         {:delete, name},
-        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
-          state
+        %State{
+          jobs: jobs,
+          storage: storage,
+          storage_pid: storage_pid,
+          debug_logging: debug_logging
+        } = state
       ) do
     debug_logging &&
       Logger.debug(fn ->
@@ -110,12 +125,12 @@ defmodule Quantum.JobBroadcaster do
 
     case Map.fetch(jobs, name) do
       {:ok, %{state: :active}} ->
-        :ok = storage.delete_job(scheduler, name)
+        :ok = storage.delete_job(storage_pid, name)
 
         {:noreply, [{:remove, name}], %{state | jobs: Map.delete(jobs, name)}}
 
       {:ok, %{state: :inactive}} ->
-        :ok = storage.delete_job(scheduler, name)
+        :ok = storage.delete_job(storage_pid, name)
 
         {:noreply, [], %{state | jobs: Map.delete(jobs, name)}}
 
@@ -126,8 +141,12 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:change_state, name, new_state},
-        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
-          state
+        %State{
+          jobs: jobs,
+          storage: storage,
+          storage_pid: storage_pid,
+          debug_logging: debug_logging
+        } = state
       ) do
     debug_logging &&
       Logger.debug(fn ->
@@ -144,7 +163,7 @@ defmodule Quantum.JobBroadcaster do
       {:ok, job} ->
         jobs = Map.update!(jobs, name, &Job.set_state(&1, new_state))
 
-        :ok = storage.update_job_state(scheduler, job.name, new_state)
+        :ok = storage.update_job_state(storage_pid, job.name, new_state)
 
         case new_state do
           :active ->
@@ -158,8 +177,12 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         :delete_all,
-        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
-          state
+        %State{
+          jobs: jobs,
+          storage: storage,
+          storage_pid: storage_pid,
+          debug_logging: debug_logging
+        } = state
       ) do
     debug_logging &&
       Logger.debug(fn ->
@@ -168,7 +191,7 @@ defmodule Quantum.JobBroadcaster do
 
     messages = for {name, %Job{state: :active}} <- jobs, do: {:remove, name}
 
-    :ok = storage.purge(scheduler)
+    :ok = storage.purge(storage_pid)
 
     {:noreply, messages, %{state | jobs: %{}}}
   end
diff --git a/lib/quantum/job_broadcaster/state.ex b/lib/quantum/job_broadcaster/state.ex
index fddc54e..cecb45c 100644
--- a/lib/quantum/job_broadcaster/state.ex
+++ b/lib/quantum/job_broadcaster/state.ex
@@ -9,10 +9,11 @@ defmodule Quantum.JobBroadcaster.State do
           jobs: %{optional(Job.name()) => Job.t()},
           buffer: JobBroadcaster.event(),
           storage: Storage,
+          storage_pid: Storage.storage_pid(),
           scheduler: Scheduler,
           debug_logging: boolean()
         }
 
-  @enforce_keys [:jobs, :buffer, :storage, :scheduler, :debug_logging]
+  @enforce_keys [:jobs, :buffer, :storage, :storage_pid, :scheduler, :debug_logging]
   defstruct @enforce_keys
 end
diff --git a/lib/quantum/storage.ex b/lib/quantum/storage.ex
index 9672d48..a55dac5 100644
--- a/lib/quantum/storage.ex
+++ b/lib/quantum/storage.ex
@@ -1,36 +1,60 @@
 defmodule Quantum.Storage do
   @moduledoc """
   Bahaviour to be implemented by all Storage Adapters.
+
+  The calls to the storage are blocking, make sure they're fast to not block the job execution.
   """
 
   alias Quantum.Job
 
+  @typedoc """
+  The location of the `server`.
+
+  ### Values
+
+  * `nil` if the storage was not started
+  * `server()` if the storage was started
+
+  """
+  @type storage_pid :: nil | GenServer.server()
+
+  @doc """
+  Storage child spec
+
+  If the storage does not need a process, specify a function that returns `:ignore`.
+
+  ### Values
+
+  * `:scheduler` - The Scheduler
+
+  """
+  @callback child_spec(init_arg :: Keyword.t()) :: Supervisor.child_spec()
+
   @doc """
   Load saved jobs from storage.
 
   Returns `:not_applicable` if the storage has never received an `add_job` call or after it has been purged.
   In this case the jobs from the configuration will be loaded.
   """
-  @callback jobs(scheduler_module :: Quantum.t()) :: :not_applicable | [Job.t()]
+  @callback jobs(storage_pid :: storage_pid) ::
+              :not_applicable | [Job.t()]
 
   @doc """
   Save new job in storage.
   """
-  @callback add_job(scheduler_module :: Quantum.t(), job :: Job.t()) :: :ok
+  @callback add_job(storage_pid :: storage_pid, job :: Job.t()) ::
+              :ok
 
   @doc """
   Delete new job in storage.
   """
-  @callback delete_job(scheduler_module :: Quantum.t(), job :: Job.name()) :: :ok
+  @callback delete_job(storage_pid :: storage_pid, job :: Job.name()) :: :ok
 
   @doc """
   Change Job State from given job.
   """
-  @callback update_job_state(
-              scheduler_module :: Quantum.t(),
-              job :: Job.name(),
-              state :: Job.state()
-            ) :: :ok
+  @callback update_job_state(storage_pid :: storage_pid, job :: Job.name(), state :: Job.state()) ::
+              :ok
 
   @doc """
   Load last execution time from storage.
@@ -38,18 +62,18 @@ defmodule Quantum.Storage do
   Returns `:unknown` if the storage does not know the last execution time.
   In this case all jobs will be run at the next applicable date.
   """
-  @callback last_execution_date(scheduler_module :: Quantum.t()) :: :unknown | NaiveDateTime.t()
+  @callback last_execution_date(storage_pid :: storage_pid) :: :unknown | NaiveDateTime.t()
 
   @doc """
   Update last execution time to given date.
   """
   @callback update_last_execution_date(
-              scheduler_module :: Quantum.t(),
+              storage_pid :: storage_pid,
               last_execution_date :: NaiveDateTime.t()
             ) :: :ok
 
   @doc """
   Purge all date from storage and go back to initial state.
   """
-  @callback purge(scheduler_module :: Quantum.t()) :: :ok
+  @callback purge(storage_pid :: storage_pid) :: :ok
 end
diff --git a/lib/quantum/storage/noop.ex b/lib/quantum/storage/noop.ex
index f546865..775d7a3 100644
--- a/lib/quantum/storage/noop.ex
+++ b/lib/quantum/storage/noop.ex
@@ -5,24 +5,40 @@ defmodule Quantum.Storage.Noop do
 
   @behaviour Quantum.Storage
 
+  use GenServer
+
+  @doc false
+  @impl GenServer
+  def init(_args), do: {:ok, nil}
+
+  @doc false
+  def start_link(_opts), do: :ignore
+
+  @doc false
   @impl Quantum.Storage
-  def jobs(_scheduler_module), do: :not_applicable
+  def jobs(_storage_pid), do: :not_applicable
 
+  @doc false
   @impl Quantum.Storage
-  def add_job(_scheduler_module, _job), do: :ok
+  def add_job(_storage_pid, _job), do: :ok
 
+  @doc false
   @impl Quantum.Storage
-  def delete_job(_scheduler_module, _job_name), do: :ok
+  def delete_job(_storage_pid, _job_name), do: :ok
 
+  @doc false
   @impl Quantum.Storage
-  def update_job_state(_scheduler_module, _job_name, _state), do: :ok
+  def update_job_state(_storage_pid, _job_name, _state), do: :ok
 
+  @doc false
   @impl Quantum.Storage
-  def last_execution_date(_scheduler_module), do: :unknown
+  def last_execution_date(_storage_pid), do: :unknown
 
+  @doc false
   @impl Quantum.Storage
-  def update_last_execution_date(_scheduler_module, _last_execution_date), do: :ok
+  def update_last_execution_date(_storage_pid, _last_execution_date), do: :ok
 
+  @doc false
   @impl Quantum.Storage
-  def purge(_scheduler_module), do: :ok
+  def purge(_storage_pid), do: :ok
 end
diff --git a/lib/quantum/supervisor.ex b/lib/quantum/supervisor.ex
index e3e3892..071a821 100644
--- a/lib/quantum/supervisor.ex
+++ b/lib/quantum/supervisor.ex
@@ -15,27 +15,28 @@ defmodule Quantum.Supervisor do
   end
 
   @impl Supervisor
-  def init({quantum, otp_app, opts}) do
-    opts = runtime_config(quantum, otp_app, opts)
-    opts = quantum_init(quantum, opts)
-    %{storage: storage, scheduler: scheduler, quantum: quantum} = opts = Map.new(opts)
+  def init({scheduler, otp_app, opts}) do
+    opts = runtime_config(scheduler, otp_app, opts)
+    opts = quantum_init(scheduler, opts)
+    %{storage: storage, scheduler: ^scheduler} = opts = Map.new(opts)
+
+    storage_opts =
+      opts
+      |> Map.get(:storage_opts, [])
+      |> Keyword.put(:scheduler, scheduler)
 
     task_registry_opts = %Quantum.TaskRegistry.StartOpts{
-      name: Module.concat(quantum, TaskRegistry)
+      name: Module.concat(scheduler, TaskRegistry)
     }
 
     clock_broadcaster_opts =
       struct!(
         Quantum.ClockBroadcaster.StartOpts,
         opts
-        |> Map.take([:debug_logging])
+        |> Map.take([:debug_logging, :storage, :scheduler])
         |> Map.merge(%{
-          name: Module.concat(quantum, ClockBroadcaster),
-          start_time:
-            case storage.last_execution_date(scheduler) do
-              :unknown -> NaiveDateTime.utc_now()
-              date -> date
-            end
+          name: Module.concat(scheduler, ClockBroadcaster),
+          start_time: NaiveDateTime.utc_now()
         })
       )
 
@@ -45,7 +46,7 @@ defmodule Quantum.Supervisor do
         opts
         |> Map.take([:jobs, :storage, :scheduler, :debug_logging])
         |> Map.merge(%{
-          name: Module.concat(quantum, JobBroadcaster)
+          name: Module.concat(scheduler, JobBroadcaster)
         })
       )
 
@@ -59,16 +60,16 @@ defmodule Quantum.Supervisor do
           :debug_logging
         ])
         |> Map.merge(%{
-          job_broadcaster_reference: Module.concat(quantum, JobBroadcaster),
-          clock_broadcaster_reference: Module.concat(quantum, ClockBroadcaster),
-          name: Module.concat(quantum, ExecutionBroadcaster)
+          job_broadcaster_reference: Module.concat(scheduler, JobBroadcaster),
+          clock_broadcaster_reference: Module.concat(scheduler, ClockBroadcaster),
+          name: Module.concat(scheduler, ExecutionBroadcaster)
         })
       )
 
     node_selector_broadcaster_opts = %Quantum.NodeSelectorBroadcaster.StartOpts{
-      execution_broadcaster_reference: Module.concat(quantum, ExecutionBroadcaster),
-      task_supervisor_reference: Module.concat(quantum, TaskSupervisor),
-      name: Module.concat(quantum, NodeSelectorBroadcaster)
+      execution_broadcaster_reference: Module.concat(scheduler, ExecutionBroadcaster),
+      task_supervisor_reference: Module.concat(scheduler, TaskSupervisor),
+      name: Module.concat(scheduler, NodeSelectorBroadcaster)
     }
 
     executor_supervisor_opts =
@@ -77,16 +78,17 @@ defmodule Quantum.Supervisor do
         opts
         |> Map.take([:debug_logging])
         |> Map.merge(%{
-          node_selector_broadcaster_reference: Module.concat(quantum, NodeSelectorBroadcaster),
-          task_supervisor_reference: Module.concat(quantum, TaskSupervisor),
-          task_registry_reference: Module.concat(quantum, TaskRegistry),
-          name: Module.concat(quantum, ExecutorSupervisor)
+          node_selector_broadcaster_reference: Module.concat(scheduler, NodeSelectorBroadcaster),
+          task_supervisor_reference: Module.concat(scheduler, TaskSupervisor),
+          task_registry_reference: Module.concat(scheduler, TaskRegistry),
+          name: Module.concat(scheduler, ExecutorSupervisor)
         })
       )
 
     Supervisor.init(
       [
-        {Task.Supervisor, [name: Module.concat(quantum, TaskSupervisor)]},
+        {Task.Supervisor, [name: Module.concat(scheduler, TaskSupervisor)]},
+        {storage, storage_opts ++ [name: Module.concat(scheduler, Storage)]},
         {Quantum.ClockBroadcaster, clock_broadcaster_opts},
         {Quantum.TaskRegistry, task_registry_opts},
         {Quantum.JobBroadcaster, job_broadcaster_opts},


Test patch:
diff --git a/test/quantum/clock_broadcaster_test.exs b/test/quantum/clock_broadcaster_test.exs
index 2db4978..bea22b4 100644
--- a/test/quantum/clock_broadcaster_test.exs
+++ b/test/quantum/clock_broadcaster_test.exs
@@ -129,7 +129,9 @@ defmodule Quantum.ClockBroadcasterTest do
            [
              name: Module.concat(__MODULE__, test),
              debug_logging: false,
-             start_time: NaiveDateTime.utc_now()
+             start_time: NaiveDateTime.utc_now(),
+             storage: Quantum.Storage.Test,
+             scheduler: NotNeeded
            ],
            Enum.to_list(opts)
          )
diff --git a/test/quantum/execution_broadcaster_test.exs b/test/quantum/execution_broadcaster_test.exs
index ed4f3ed..523e7d0 100644
--- a/test/quantum/execution_broadcaster_test.exs
+++ b/test/quantum/execution_broadcaster_test.exs
@@ -113,8 +113,7 @@ defmodule Quantum.ExecutionBroadcasterTest do
         now = %{NaiveDateTime.utc_now() | microsecond: {0, 0}}
         TestProducer.send(producer, %ClockEvent{time: now, catch_up: false})
 
-        assert_receive {:update_last_execution_date, {TestScheduler, %NaiveDateTime{}}, _},
-                       @max_timeout
+        assert_receive {:update_last_execution_date, %NaiveDateTime{}, _}, @max_timeout
 
         assert_receive {:received, %ExecuteEvent{job: ^job}}, @max_timeout
       end)
diff --git a/test/quantum/job_broadcaster_test.exs b/test/quantum/job_broadcaster_test.exs
index 3d7fa90..c1c0abc 100644
--- a/test/quantum/job_broadcaster_test.exs
+++ b/test/quantum/job_broadcaster_test.exs
@@ -123,7 +123,7 @@ defmodule Quantum.JobBroadcasterTest do
 
                assert_receive {:received, {:add, ^active_job}}
 
-               assert_receive {:add_job, {TestScheduler, ^active_job}, _}
+               assert_receive {:add_job, ^active_job, _}
              end) =~ "Adding job #Reference"
     end
 
@@ -160,7 +160,7 @@ defmodule Quantum.JobBroadcasterTest do
 
         refute_receive {:received, {:add, _}}
 
-        assert_receive {:add_job, {TestScheduler, ^inactive_job}, _}
+        assert_receive {:add_job, ^inactive_job, _}
       end)
     end
   end
@@ -175,7 +175,7 @@ defmodule Quantum.JobBroadcasterTest do
 
         assert_receive {:received, {:remove, ^active_job_name}}
 
-        assert_receive {:delete_job, {TestScheduler, ^active_job_name}, _}
+        assert_receive {:delete_job, ^active_job_name, _}
 
         refute Enum.any?(TestScheduler.jobs(broadcaster), fn {key, _} ->
                  key == active_job_name
@@ -203,7 +203,7 @@ defmodule Quantum.JobBroadcasterTest do
 
         refute_receive {:received, {:remove, _}}
 
-        assert_receive {:delete_job, {TestScheduler, ^inactive_job_name}, _}
+        assert_receive {:delete_job, ^inactive_job_name, _}
 
         refute Enum.any?(TestScheduler.jobs(broadcaster), fn {key, _} ->
                  key == inactive_job.name
@@ -222,7 +222,7 @@ defmodule Quantum.JobBroadcasterTest do
 
         assert_receive {:received, {:remove, ^active_job_name}}
 
-        assert_receive {:update_job_state, {TestScheduler, _, _}, _}
+        assert_receive {:update_job_state, {_, _}, _}
       end)
     end
 
@@ -235,7 +235,7 @@ defmodule Quantum.JobBroadcasterTest do
 
         assert_receive {:received, {:add, ^active_job}}
 
-        assert_receive {:update_job_state, {TestScheduler, _, _}, _}
+        assert_receive {:update_job_state, {_, _}, _}
       end)
     end
 
@@ -295,7 +295,7 @@ defmodule Quantum.JobBroadcasterTest do
         refute_receive {:received, {:remove, ^inactive_job_name}}
         assert_receive {:received, {:remove, ^active_job_name}}
 
-        assert_receive {:purge, TestScheduler, _}
+        assert_receive {:purge, _, _}
       end)
     end
   end
diff --git a/test/support/test_storage.ex b/test/support/test_storage.ex
index 6f037fc..a4c6934 100644
--- a/test/support/test_storage.ex
+++ b/test/support/test_storage.ex
@@ -5,49 +5,75 @@ defmodule Quantum.Storage.Test do
 
   @behaviour Quantum.Storage
 
-  def jobs(scheduler_module), do: send_and_wait(:jobs, scheduler_module, :not_applicable)
-  def add_job(scheduler_module, job), do: send_and_wait(:add_job, {scheduler_module, job})
+  use GenServer
 
-  def delete_job(scheduler_module, job_name),
-    do: send_and_wait(:delete_job, {scheduler_module, job_name})
+  def start_link(_opts), do: send_and_wait(:jobs, :start_link, :ignore)
 
-  def update_job_state(scheduler_module, job_name, state),
-    do: send_and_wait(:update_job_state, {scheduler_module, job_name, state})
+  @doc false
+  @impl GenServer
+  def init(_args), do: {:ok, nil}
 
-  def last_execution_date(scheduler_module),
-    do: send_and_wait(:last_execution_date, scheduler_module, :unknown)
+  @impl Quantum.Storage
+  def jobs(_storage_pid), do: send_and_wait(:jobs, nil, :not_applicable)
 
-  def update_last_execution_date(scheduler_module, last_execution_date),
-    do: send_and_wait(:update_last_execution_date, {scheduler_module, last_execution_date})
+  @impl Quantum.Storage
+  def add_job(_storage_pid, job), do: send_and_wait(:add_job, job)
 
-  def purge(scheduler_module), do: send_and_wait(:purge, scheduler_module)
+  @impl Quantum.Storage
+  def delete_job(_storage_pid, job_name), do: send_and_wait(:delete_job, job_name)
+
+  @impl Quantum.Storage
+  def update_job_state(_storage_pid, job_name, state),
+    do: send_and_wait(:update_job_state, {job_name, state})
+
+  @impl Quantum.Storage
+  def last_execution_date(_storage_pid), do: send_and_wait(:last_execution_date, nil, :unknown)
+
+  @impl Quantum.Storage
+  def update_last_execution_date(_storage_pid, last_execution_date),
+    do: send_and_wait(:update_last_execution_date, last_execution_date)
+
+  @impl Quantum.Storage
+  def purge(_storage_pid), do: send_and_wait(:purge, nil)
 
   # Used for Small Test Storages
   defmacro __using__(_) do
     quote do
       @behaviour Quantum.Storage
 
-      alias Quantum.Storage.Test
+      import Quantum.Storage.Test
+
+      use GenServer
+
+      def start_link(_opts), do: send_and_wait(:jobs, :start_link, :ignore)
+
+      @doc false
+      @impl GenServer
+      def init(_args), do: {:ok, nil}
 
-      def jobs(scheduler_module), do: Test.send_and_wait(:jobs, scheduler_module, :not_applicable)
+      @impl Quantum.Storage
+      def jobs(_storage_pid), do: send_and_wait(:jobs, nil, :not_applicable)
 
-      def add_job(scheduler_module, job),
-        do: Test.send_and_wait(:add_job, {scheduler_module, job})
+      @impl Quantum.Storage
+      def add_job(_storage_pid, job), do: send_and_wait(:add_job, job)
 
-      def delete_job(scheduler_module, job_name),
-        do: Test.send_and_wait(:delete_job, {scheduler_module, job_name})
+      @impl Quantum.Storage
+      def delete_job(_storage_pid, job_name), do: send_and_wait(:delete_job, job_name)
 
-      def update_job_state(scheduler_module, job_name, state),
-        do: Test.send_and_wait(:update_job_state, {scheduler_module, job_name, state})
+      @impl Quantum.Storage
+      def update_job_state(_storage_pid, job_name, state),
+        do: send_and_wait(:update_job_state, job_name, state)
 
-      def last_execution_date(scheduler_module),
-        do: Test.send_and_wait(:last_execution_date, scheduler_module, :unknown)
+      @impl Quantum.Storage
+      def last_execution_date(_storage_pid),
+        do: send_and_wait(:last_execution_date, nil, :unknown)
 
-      def update_last_execution_date(scheduler_module, last_execution_date),
-        do:
-          Test.send_and_wait(:update_last_execution_date, {scheduler_module, last_execution_date})
+      @impl Quantum.Storage
+      def update_last_execution_date(_storage_pid, last_execution_date),
+        do: send_and_wait(:update_last_execution_date, last_execution_date)
 
-      def purge(scheduler_module), do: Test.send_and_wait(:purge, scheduler_module)
+      @impl Quantum.Storage
+      def purge(_storage_pid), do: send_and_wait(:purge, nil)
 
       defoverridable Quantum.Storage
     end