Repository: quantum-elixir/quantum-core
Pull request #313 (2018-02-28 15:30:19)

Solution: Storage Adapter
~~Based on #317, merge that PR first.~~

Partial Implementation of #240

This PR is currently only to track changes, do not merge until it is proven to be working with an actual storage implementation.

c-rack: @maennchen transfer completed, fork is already renamed and I've added you as collaborator for the README changes.
maennchen: @pyatkov I removed your changes and put them into this repo: https://github.com/quantum-elixir/quantum-storage-ets

I will stabilize the code and add tests. After that we should be able to release the new API and the ets storage package.
coveralls: ## Pull Request Test Coverage Report for [Build 767](https://coveralls.io/builds/16139269)

* **49** of **49**  **(100.0%)** changed or added relevant lines in **5** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**2.4%**) to **94.481%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/16139269/badge)](https://coveralls.io/builds/16139269) |
| :-- | --: |
| Change from base [Build 764](https://coveralls.io/builds/16128563): |  2.4% |
| Covered Lines: | 291 |
| Relevant Lines: | 308 |

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

maennchen: @c-rack Would it be ok to mark this module as "experimental" and not adhere to the semantic versioning as we do with the rest of the library. (Only until it is stabilized)
maennchen: @c-rack Great! Then I'm ready to release it into the wild I think.
maennchen: PS: I'm adding documentation as soon as it is stable.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index feed9d9..eb01589 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,9 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 
+### Added
+- Experimental Storage API
+
 Diff for [unreleased]
 
 ## 2.2.7 - 2018-03-22
diff --git a/lib/quantum.ex b/lib/quantum.ex
index a1dfd81..7e2a667 100644
--- a/lib/quantum.ex
+++ b/lib/quantum.ex
@@ -8,6 +8,7 @@ defmodule Quantum do
   alias Quantum.Normalizer
   alias Quantum.Job
   alias Quantum.RunStrategy.Random
+  alias Quantum.Storage.Noop
 
   @defaults [
     global: false,
@@ -46,6 +47,7 @@ defmodule Quantum do
     task_supervisor = Module.concat(quantum, Task.Supervisor)
 
     config
+    |> Keyword.put_new(:quantum, quantum)
     |> update_in([:schedule], &Normalizer.normalize_schedule/1)
     |> Keyword.put_new(:task_stages_supervisor, task_stages_supervisor)
     |> Keyword.put_new(:job_broadcaster, job_broadcaster)
@@ -53,6 +55,7 @@ defmodule Quantum do
     |> Keyword.put_new(:executor_supervisor, executor_supervisor)
     |> Keyword.put_new(:task_registry, task_registry)
     |> Keyword.put_new(:task_supervisor, task_supervisor)
+    |> Keyword.put_new(:storage, Noop)
   end
 
   @doc """
diff --git a/lib/quantum/execution_broadcaster.ex b/lib/quantum/execution_broadcaster.ex
index 85cfec8..442d4a0 100644
--- a/lib/quantum/execution_broadcaster.ex
+++ b/lib/quantum/execution_broadcaster.ex
@@ -8,8 +8,11 @@ defmodule Quantum.ExecutionBroadcaster do
   require Logger
 
   alias Quantum.{Job, Util, DateLibrary}
-  alias Crontab.{Scheduler, CronExpression}
   alias Quantum.DateLibrary.{InvalidDateTimeForTimezoneError, InvalidTimezoneError}
+  alias Crontab.Scheduler, as: CrontabScheduler
+  alias Crontab.CronExpression
+  alias Quantum.Storage.Adapter
+  alias Quantum.Scheduler
 
   defmodule JobInPastError do
     defexception message:
@@ -25,22 +28,57 @@ defmodule Quantum.ExecutionBroadcaster do
     * `job_broadcaster` - The name of the stage to listen to
 
   """
-  @spec start_link(GenServer.server(), GenServer.server(), boolean()) :: GenServer.on_start()
-  def start_link(name, job_broadcaster, debug_logging) do
+  @spec start_link(GenServer.server(), GenServer.server(), Adapter, Scheduler, boolean()) ::
+          GenServer.on_start()
+  def start_link(name, job_broadcaster, storage, scheduler, debug_logging) do
     __MODULE__
-    |> GenStage.start_link({job_broadcaster, debug_logging}, name: name)
+    |> GenStage.start_link({job_broadcaster, storage, scheduler, debug_logging}, name: name)
     |> Util.start_or_link()
   end
 
   @doc false
-  @spec child_spec({GenServer.server(), GenServer.server(), boolean()}) :: Supervisor.child_spec()
-  def child_spec({name, job_broadcaster, debug_logging}) do
-    %{super([]) | start: {__MODULE__, :start_link, [name, job_broadcaster, debug_logging]}}
+  @spec child_spec({GenServer.server(), GenServer.server(), Adapter, Scheduler, boolean()}) ::
+          Supervisor.child_spec()
+  def child_spec({name, job_broadcaster, storage, scheduler, debug_logging}) do
+    %{
+      super([])
+      | start:
+          {__MODULE__, :start_link, [name, job_broadcaster, storage, scheduler, debug_logging]}
+    }
   end
 
   @doc false
-  def init({job_broadcaster, debug_logging}) do
-    state = %{jobs: [], time: NaiveDateTime.utc_now(), timer: nil, debug_logging: debug_logging}
+  def init({job_broadcaster, storage, scheduler, debug_logging}) do
+    last_execution_date =
+      case storage.last_execution_date(scheduler) do
+        %NaiveDateTime{} = date ->
+          debug_logging &&
+            Logger.debug(fn ->
+              "[#{inspect(Node.self())}][#{__MODULE__}] Using last known execution time #{
+                NaiveDateTime.to_iso8601(date)
+              }"
+            end)
+
+          date
+
+        :unknown ->
+          debug_logging &&
+            Logger.debug(fn ->
+              "[#{inspect(Node.self())}][#{__MODULE__}] Unknown last execution time, using now"
+            end)
+
+          NaiveDateTime.utc_now()
+      end
+
+    state = %{
+      jobs: [],
+      time: last_execution_date,
+      timer: nil,
+      storage: storage,
+      scheduler: scheduler,
+      debug_logging: debug_logging
+    }
+
     {:producer_consumer, state, subscribe_to: [job_broadcaster]}
   end
 
@@ -71,8 +109,15 @@ defmodule Quantum.ExecutionBroadcaster do
 
   def handle_info(
         :execute,
-        %{jobs: [{time_to_execute, jobs_to_execute} | tail], debug_logging: debug_logging} = state
+        %{
+          jobs: [{time_to_execute, jobs_to_execute} | tail],
+          storage: storage,
+          scheduler: scheduler,
+          debug_logging: debug_logging
+        } = state
       ) do
+    :ok = storage.update_last_execution_date(scheduler, time_to_execute)
+
     state =
       state
       |> Map.put(:timer, nil)
@@ -126,6 +171,8 @@ defmodule Quantum.ExecutionBroadcaster do
       end)
 
     %{state | jobs: jobs}
+    |> sort_state
+    |> reset_timer
   end
 
   defp add_job_to_state(
@@ -162,7 +209,7 @@ defmodule Quantum.ExecutionBroadcaster do
          time
        ) do
     schedule
-    |> Scheduler.get_next_run_date(DateLibrary.to_tz!(time, timezone))
+    |> CrontabScheduler.get_next_run_date(DateLibrary.to_tz!(time, timezone))
     |> case do
       {:ok, date} ->
         {:ok, DateLibrary.to_utc!(date, timezone)}
diff --git a/lib/quantum/job.ex b/lib/quantum/job.ex
index 4c26a22..628d395 100644
--- a/lib/quantum/job.ex
+++ b/lib/quantum/job.ex
@@ -25,13 +25,14 @@ defmodule Quantum.Job do
     state: :active
   ]
 
+  @type name :: atom | reference()
   @type state :: :active | :inactive
   @type task :: {atom, atom, [any]} | (() -> any)
   @type timezone :: :utc | :local | String.t()
   @type schedule :: Crontab.CronExpression.t()
 
   @type t :: %__MODULE__{
-          name: atom | Reference,
+          name: name,
           schedule: schedule | nil,
           task: task | nil,
           state: state,
diff --git a/lib/quantum/job_broadcaster.ex b/lib/quantum/job_broadcaster.ex
index c8c716b..3528ac0 100644
--- a/lib/quantum/job_broadcaster.ex
+++ b/lib/quantum/job_broadcaster.ex
@@ -7,7 +7,8 @@ defmodule Quantum.JobBroadcaster do
 
   require Logger
 
-  alias Quantum.{Job, Util}
+  alias Quantum.{Job, Util, Scheduler}
+  alias Quantum.Storage.Adapter
 
   @doc """
   Start Job Broadcaster
@@ -18,24 +19,52 @@ defmodule Quantum.JobBroadcaster do
    * `jobs` - Array of `Quantum.Job`
 
   """
-  @spec start_link(GenServer.server(), [Job.t()], boolean()) :: GenServer.on_start()
-  def start_link(name, jobs, debug_logging) do
+  @spec start_link(GenServer.server(), [Job.t()], Adapter, Scheduler, boolean()) ::
+          GenServer.on_start()
+  def start_link(name, jobs, storage, scheduler, debug_logging) do
     __MODULE__
-    |> GenStage.start_link({jobs, debug_logging}, name: name)
+    |> GenStage.start_link({jobs, storage, scheduler, debug_logging}, name: name)
     |> Util.start_or_link()
   end
 
   @doc false
-  @spec child_spec({GenServer.server(), [Job.t()], boolean()}) :: Supervisor.child_spec()
-  def child_spec({name, jobs, debug_logging}) do
-    %{super([]) | start: {__MODULE__, :start_link, [name, jobs, debug_logging]}}
+  @spec child_spec({GenServer.server(), [Job.t()], Adapter, Scheduler, boolean()}) ::
+          Supervisor.child_spec()
+  def child_spec({name, jobs, storage, scheduler, debug_logging}) do
+    %{
+      super([])
+      | start: {__MODULE__, :start_link, [name, jobs, storage, scheduler, debug_logging]}
+    }
   end
 
   @doc false
-  def init({jobs, debug_logging}) do
+  def init({jobs, storage, scheduler, debug_logging}) do
+    effective_jobs =
+      scheduler
+      |> storage.jobs()
+      |> case do
+        :not_applicable ->
+          debug_logging &&
+            Logger.debug(fn ->
+              "[#{inspect(Node.self())}][#{__MODULE__}] Loading Initial Jobs from Config"
+            end)
+
+          jobs
+
+        storage_jobs when is_list(storage_jobs) ->
+          debug_logging &&
+            Logger.debug(fn ->
+              "[#{inspect(Node.self())}][#{__MODULE__}] Loading Initial Jobs from Storage, skipping config"
+            end)
+
+          storage_jobs
+      end
+
     state = %{
-      jobs: Enum.into(jobs, %{}, fn %{name: name} = job -> {name, job} end),
-      buffer: for(%{state: :active} = job <- jobs, do: {:add, job}),
+      jobs: Enum.into(effective_jobs, %{}, fn %{name: name} = job -> {name, job} end),
+      buffer: for(%{state: :active} = job <- effective_jobs, do: {:add, job}),
+      storage: storage,
+      scheduler: scheduler,
       debug_logging: debug_logging
     }
 
@@ -50,29 +79,39 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:add, %Job{state: :active, name: job_name} = job},
-        %{jobs: jobs, debug_logging: debug_logging} = state
+        %{jobs: jobs, storage: storage, scheduler: scheduler, 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)
+
     {:noreply, [{:add, job}], %{state | jobs: Map.put(jobs, job_name, job)}}
   end
 
   def handle_cast(
         {:add, %Job{state: :inactive, name: job_name} = job},
-        %{jobs: jobs, debug_logging: debug_logging} = state
+        %{jobs: jobs, storage: storage, scheduler: scheduler, 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)
+
     {:noreply, [], %{state | jobs: Map.put(jobs, job_name, job)}}
   end
 
-  def handle_cast({:delete, name}, %{jobs: jobs, debug_logging: debug_logging} = state) do
+  def handle_cast(
+        {:delete, name},
+        %{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
+          state
+      ) do
     debug_logging &&
       Logger.debug(fn ->
         "[#{inspect(Node.self())}][#{__MODULE__}] Deleting job #{inspect(name)}"
@@ -80,9 +119,13 @@ defmodule Quantum.JobBroadcaster do
 
     case Map.fetch(jobs, name) do
       {:ok, %{state: :active}} ->
+        :ok = storage.delete_job(scheduler, name)
+
         {:noreply, [{:remove, name}], %{state | jobs: Map.delete(jobs, name)}}
 
       {:ok, %{state: :inactive}} ->
+        :ok = storage.delete_job(scheduler, name)
+
         {:noreply, [], %{state | jobs: Map.delete(jobs, name)}}
 
       :error ->
@@ -92,7 +135,8 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:change_state, name, new_state},
-        %{jobs: jobs, debug_logging: debug_logging} = state
+        %{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
+          state
       ) do
     debug_logging &&
       Logger.debug(fn ->
@@ -109,6 +153,8 @@ 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)
+
         case new_state do
           :active ->
             {:noreply, [{:add, %{job | state: new_state}}], %{state | jobs: jobs}}
@@ -119,7 +165,11 @@ defmodule Quantum.JobBroadcaster do
     end
   end
 
-  def handle_cast(:delete_all, %{jobs: jobs, debug_logging: debug_logging} = state) do
+  def handle_cast(
+        :delete_all,
+        %{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
+          state
+      ) do
     debug_logging &&
       Logger.debug(fn ->
         "[#{inspect(Node.self())}][#{__MODULE__}] Deleting all jobs"
@@ -127,6 +177,8 @@ defmodule Quantum.JobBroadcaster do
 
     messages = for {name, %Job{state: :active}} <- jobs, do: {:remove, name}
 
+    :ok = storage.purge(scheduler)
+
     {:noreply, messages, %{state | jobs: %{}}}
   end
 
diff --git a/lib/quantum/storage/adapter.ex b/lib/quantum/storage/adapter.ex
new file mode 100644
index 0000000..e538f8b
--- /dev/null
+++ b/lib/quantum/storage/adapter.ex
@@ -0,0 +1,64 @@
+defmodule Quantum.Storage.Adapter do
+  @moduledoc """
+  Bahaviour to be implemented by all Storage Adapters.
+
+  **WARNING: This Adapter is experimental and will therefore not adhere to semantic versioning.
+  It could undergo massive changes even in patch releases.**
+  """
+
+  alias Quantum.Job
+
+  @typedoc """
+  The calling scheduler Module
+  """
+  @type scheduler_module :: atom
+
+  @typedoc """
+  The expected return is `:ok`, every other result will terminate the scheduler.
+  """
+  @type ok :: :ok
+
+  @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 weill be loaded.
+  """
+  @callback jobs(scheduler_module) :: :not_applicable | [Job.t()]
+
+  @doc """
+  Save new job in storage.
+  """
+  @callback add_job(scheduler_module, job :: Job.t()) :: ok
+
+  @doc """
+  Delete new job in storage.
+  """
+  @callback delete_job(scheduler_module, job :: Job.name()) :: ok
+
+  @doc """
+  Change Job State from given job.
+  """
+  @callback update_job_state(scheduler_module, job :: Job.name(), state :: Job.state()) :: ok
+
+  @doc """
+  Load last execution time from storage
+
+  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) :: :unknown | NaiveDateTime.t()
+
+  @doc """
+  Update last execution time to given date.
+  """
+  @callback update_last_execution_date(
+              scheduler_module,
+              last_execution_date :: NaiveDateTime.t()
+            ) :: ok
+
+  @doc """
+  Purge all date from storage and go back to initial state.
+  """
+  @callback purge(scheduler_module) :: ok
+end
diff --git a/lib/quantum/storage/noop.ex b/lib/quantum/storage/noop.ex
new file mode 100644
index 0000000..a87df3b
--- /dev/null
+++ b/lib/quantum/storage/noop.ex
@@ -0,0 +1,15 @@
+defmodule Quantum.Storage.Noop do
+  @moduledoc """
+  Empty implementation of a `Quantum.Storage.Adapter`.
+  """
+
+  @behaviour Quantum.Storage.Adapter
+
+  def jobs(_scheduler_module), do: :not_applicable
+  def add_job(_scheduler_module, _job), do: :ok
+  def delete_job(_scheduler_module, _job_name), do: :ok
+  def update_job_state(_scheduler_module, _job_name, _state), do: :ok
+  def last_execution_date(_scheduler_module), do: :unknown
+  def update_last_execution_date(_scheduler_module, _last_execution_date), do: :ok
+  def purge(_scheduler_module), do: :ok
+end
diff --git a/lib/quantum/task_stages_supervisor.ex b/lib/quantum/task_stages_supervisor.ex
index bedd144..80d519a 100644
--- a/lib/quantum/task_stages_supervisor.ex
+++ b/lib/quantum/task_stages_supervisor.ex
@@ -33,6 +33,8 @@ defmodule Quantum.TaskStagesSupervisor do
           {
             Keyword.fetch!(opts, :job_broadcaster),
             Keyword.fetch!(opts, :jobs),
+            Keyword.fetch!(opts, :storage),
+            Keyword.fetch!(opts, :quantum),
             Keyword.fetch!(opts, :debug_logging)
           }
         },
@@ -41,6 +43,8 @@ defmodule Quantum.TaskStagesSupervisor do
           {
             Keyword.fetch!(opts, :execution_broadcaster),
             Keyword.fetch!(opts, :job_broadcaster),
+            Keyword.fetch!(opts, :storage),
+            Keyword.fetch!(opts, :quantum),
             Keyword.fetch!(opts, :debug_logging)
           }
         },
diff --git a/pages/configuration.md b/pages/configuration.md
index 075016e..269efb6 100644
--- a/pages/configuration.md
+++ b/pages/configuration.md
@@ -19,6 +19,25 @@ config :your_app, YourApp.Scheduler,
   ]
 ```
 
+## Persistent Storage
+
+Persistent storage can be used to track jobs and last execution times over restarts.
+
+**Note: If a storage is present, the jobs from the configuration will not be loaded to prevent conflicts.**
+
+```elixir
+config :your_app, YourApp.Scheduler,
+  storage: Quantum.Storage.Adapter.Implementation
+```
+
+### Storage Adapters
+
+Storage implementations must implement the `Quantum.Storage.Adapter` behaviour.
+
+The following adapters are supported:
+
+* TODO: Add Adapters
+
 ## Release managers
 (
 [conform](https://github.com/bitwalker/conform) /


Test patch:
diff --git a/test/quantum/execution_broadcaster_test.exs b/test/quantum/execution_broadcaster_test.exs
index dcd6f37..d118492 100644
--- a/test/quantum/execution_broadcaster_test.exs
+++ b/test/quantum/execution_broadcaster_test.exs
@@ -5,10 +5,12 @@ defmodule Quantum.ExecutionBroadcasterTest do
 
   import Crontab.CronExpression
   import ExUnit.CaptureLog
+  import Quantum.CaptureLogExtend
 
   alias Quantum.ExecutionBroadcaster
   alias Quantum.{TestConsumer, TestProducer}
   alias Quantum.Job
+  alias Quantum.Storage.Test, as: TestStorage
 
   # Allow max 10% Latency
   @max_timeout 1_100
@@ -21,12 +23,27 @@ defmodule Quantum.ExecutionBroadcasterTest do
     use Quantum.Scheduler, otp_app: :execution_broadcaster_test
   end
 
-  setup do
-    {:ok, producer} = start_supervised({TestProducer, []})
-    {:ok, broadcaster} = start_supervised({ExecutionBroadcaster, {__MODULE__, producer, true}})
-    {:ok, _consumer} = start_supervised({TestConsumer, [broadcaster, self()]})
+  setup tags do
+    if tags[:listen_storage] do
+      Process.put(:test_pid, self())
+    end
+
+    if tags[:manual_dispatch] do
+      :ok
+    else
+      {:ok, producer} = start_supervised({TestProducer, []})
+
+      {{:ok, broadcaster}, _} =
+        capture_log_with_return(fn ->
+          start_supervised(
+            {ExecutionBroadcaster, {__MODULE__, producer, TestStorage, TestScheduler, true}}
+          )
+        end)
 
-    {:ok, %{producer: producer, broadcaster: broadcaster, debug_logging: true}}
+      {:ok, _consumer} = start_supervised({TestConsumer, [broadcaster, self()]})
+
+      {:ok, %{producer: producer, broadcaster: broadcaster, debug_logging: true}}
+    end
   end
 
   describe "add" do
@@ -62,6 +79,71 @@ defmodule Quantum.ExecutionBroadcasterTest do
       end)
     end
 
+    @tag manual_dispatch: true, listen_storage: true
+    test "loads last execution time from storage" do
+      defmodule TestStorageWithLastExecutionTime do
+        @moduledoc false
+        use Quantum.Storage.Test
+
+        def last_execution_date(_),
+          do: NaiveDateTime.add(NaiveDateTime.utc_now(), -3_600, :second)
+      end
+
+      capture_log(fn ->
+        {:ok, producer} = start_supervised({TestProducer, []})
+
+        {:ok, broadcaster} =
+          start_supervised(
+            {ExecutionBroadcaster,
+             {__MODULE__, producer, TestStorageWithLastExecutionTime, TestScheduler, true}}
+          )
+
+        {:ok, _consumer} = start_supervised({TestConsumer, [broadcaster, self()]})
+
+        job =
+          TestScheduler.new_job()
+          |> Job.set_schedule(~e[*]e)
+
+        TestProducer.send(producer, {:add, job})
+
+        assert_receive {:update_last_execution_date, {TestScheduler, date}, _}, @max_timeout
+
+        diff_seconds = NaiveDateTime.diff(NaiveDateTime.utc_now(), date, :second)
+
+        assert diff_seconds >= 3_600 - 1
+
+        assert_receive {:received, {:execute, ^job}}, @max_timeout
+        # Quickly executes until reached current time
+        for _ <- 0..diff_seconds do
+          assert_receive {:received, {:execute, ^job}}, 100
+        end
+
+        # Maybe a little time elapsed in the test?
+        for _ <- 0..2 do
+          assert_receive {:received, {:execute, ^job}}, 1010
+        end
+
+        # Goes back to normal pace
+        refute_receive {:received, {:execute, ^job}}, 100
+      end)
+    end
+
+    @tag listen_storage: true
+    test "saves new last execution time in storage", %{producer: producer} do
+      job =
+        TestScheduler.new_job()
+        |> Job.set_schedule(~e[*]e)
+
+      capture_log(fn ->
+        TestProducer.send(producer, {:add, job})
+
+        assert_receive {:update_last_execution_date, {TestScheduler, %NaiveDateTime{}}, _},
+                       @max_timeout
+
+        assert_receive {:received, {:execute, ^job}}, @max_timeout
+      end)
+    end
+
     test "normal schedule in other timezone triggers once per second", %{producer: producer} do
       job =
         TestScheduler.new_job()
diff --git a/test/quantum/job_broadcaster_test.exs b/test/quantum/job_broadcaster_test.exs
index c1c0d55..18caa1a 100644
--- a/test/quantum/job_broadcaster_test.exs
+++ b/test/quantum/job_broadcaster_test.exs
@@ -6,8 +6,10 @@ defmodule Quantum.JobBroadcasterTest do
   alias Quantum.JobBroadcaster
   alias Quantum.TestConsumer
   alias Quantum.Job
+  alias Quantum.Storage.Test, as: TestStorage
 
   import ExUnit.CaptureLog
+  import Quantum.CaptureLogExtend
 
   doctest JobBroadcaster
 
@@ -18,6 +20,10 @@ defmodule Quantum.JobBroadcasterTest do
   end
 
   setup tags do
+    if tags[:listen_storage] do
+      Process.put(:test_pid, self())
+    end
+
     active_job = TestScheduler.new_job()
     inactive_job = Job.set_state(TestScheduler.new_job(), :inactive)
 
@@ -36,8 +42,21 @@ defmodule Quantum.JobBroadcasterTest do
           []
       end
 
-    {:ok, broadcaster} = start_supervised({JobBroadcaster, {__MODULE__, init_jobs, true}})
-    {:ok, _consumer} = start_supervised({TestConsumer, [broadcaster, self()]})
+    broadcaster =
+      if tags[:manual_dispatch] do
+        nil
+      else
+        {{:ok, broadcaster}, _} =
+          capture_log_with_return(fn ->
+            start_supervised(
+              {JobBroadcaster, {__MODULE__, init_jobs, TestStorage, TestScheduler, true}}
+            )
+          end)
+
+        {:ok, _consumer} = start_supervised({TestConsumer, [broadcaster, self()]})
+
+        broadcaster
+      end
 
     {
       :ok,
@@ -56,14 +75,42 @@ defmodule Quantum.JobBroadcasterTest do
       refute_receive {:received, {:add, ^inactive_job}}
       assert_receive {:received, {:add, ^active_job}}
     end
+
+    @tag manual_dispatch: true
+    test "storage jobs", %{active_job: active_job, inactive_job: inactive_job} do
+      capture_log(fn ->
+        defmodule FullStorage do
+          @moduledoc false
+
+          use Quantum.Storage.Test
+
+          def jobs(_),
+            do: [
+              TestScheduler.new_job(),
+              Job.set_state(TestScheduler.new_job(), :inactive)
+            ]
+        end
+
+        {:ok, broadcaster} =
+          start_supervised({JobBroadcaster, {__MODULE__, [], FullStorage, TestScheduler, true}})
+
+        {:ok, _consumer} = start_supervised({TestConsumer, [broadcaster, self()]})
+
+        assert_receive {:received, {:add, _}}
+        refute_receive {:received, {:add, _}}
+      end)
+    end
   end
 
   describe "add" do
+    @tag listen_storage: true
     test "active", %{broadcaster: broadcaster, active_job: active_job} do
       assert capture_log(fn ->
                TestScheduler.add_job(broadcaster, active_job)
 
                assert_receive {:received, {:add, ^active_job}}
+
+               assert_receive {:add_job, {TestScheduler, ^active_job}, _}
              end) =~ "Adding job #Reference"
     end
 
@@ -74,7 +121,9 @@ defmodule Quantum.JobBroadcasterTest do
                :ok = stop_supervised(TestConsumer)
 
                {:ok, broadcaster} =
-                 start_supervised({JobBroadcaster, {__MODULE__, init_jobs, false}})
+                 start_supervised(
+                   {JobBroadcaster, {__MODULE__, init_jobs, TestStorage, TestScheduler, false}}
+                 )
 
                {:ok, _consumer} = start_supervised({TestConsumer, [broadcaster, self()]})
 
@@ -84,17 +133,20 @@ defmodule Quantum.JobBroadcasterTest do
              end) =~ "Adding job #Reference"
     end
 
+    @tag listen_storage: true
     test "inactive", %{broadcaster: broadcaster, inactive_job: inactive_job} do
       capture_log(fn ->
         TestScheduler.add_job(broadcaster, inactive_job)
 
         refute_receive {:received, {:add, _}}
+
+        assert_receive {:add_job, {TestScheduler, ^inactive_job}, _}
       end)
     end
   end
 
   describe "delete" do
-    @tag jobs: :active
+    @tag jobs: :active, listen_storage: true
     test "active", %{broadcaster: broadcaster, active_job: active_job} do
       active_job_name = active_job.name
 
@@ -103,27 +155,36 @@ defmodule Quantum.JobBroadcasterTest do
 
         assert_receive {:received, {:remove, ^active_job_name}}
 
+        assert_receive {:delete_job, {TestScheduler, ^active_job_name}, _}
+
         refute Enum.any?(TestScheduler.jobs(broadcaster), fn {key, _} ->
                  key == active_job_name
                end)
       end)
     end
 
+    @tag listen_storage: true
     test "missing", %{broadcaster: broadcaster} do
       capture_log(fn ->
         TestScheduler.delete_job(broadcaster, make_ref())
 
         refute_receive {:received, {:remove, _}}
+
+        refute_receive {:delete_job, {TestScheduler, _}, _}
       end)
     end
 
-    @tag jobs: :inactive
+    @tag jobs: :inactive, listen_storage: true
     test "inactive", %{broadcaster: broadcaster, inactive_job: inactive_job} do
       capture_log(fn ->
+        inactive_job_name = inactive_job.name
+
         TestScheduler.delete_job(broadcaster, inactive_job.name)
 
         refute_receive {:received, {:remove, _}}
 
+        assert_receive {:delete_job, {TestScheduler, ^inactive_job_name}, _}
+
         refute Enum.any?(TestScheduler.jobs(broadcaster), fn {key, _} ->
                  key == inactive_job.name
                end)
@@ -132,7 +193,7 @@ defmodule Quantum.JobBroadcasterTest do
   end
 
   describe "change_state" do
-    @tag jobs: :active
+    @tag jobs: :active, listen_storage: true
     test "active => inactive", %{broadcaster: broadcaster, active_job: active_job} do
       active_job_name = active_job.name
 
@@ -140,10 +201,12 @@ defmodule Quantum.JobBroadcasterTest do
         TestScheduler.deactivate_job(broadcaster, active_job.name)
 
         assert_receive {:received, {:remove, ^active_job_name}}
+
+        assert_receive {:update_job_state, {TestScheduler, _, _}, _}
       end)
     end
 
-    @tag jobs: :inactive
+    @tag jobs: :inactive, listen_storage: true
     test "inactive => active", %{broadcaster: broadcaster, inactive_job: inactive_job} do
       capture_log(fn ->
         TestScheduler.activate_job(broadcaster, inactive_job.name)
@@ -151,10 +214,12 @@ defmodule Quantum.JobBroadcasterTest do
         active_job = Job.set_state(inactive_job, :active)
 
         assert_receive {:received, {:add, ^active_job}}
+
+        assert_receive {:update_job_state, {TestScheduler, _, _}, _}
       end)
     end
 
-    @tag jobs: :active
+    @tag jobs: :active, listen_storage: true
     test "active => active", %{broadcaster: broadcaster, active_job: active_job} do
       # Initial
       assert_receive {:received, {:add, ^active_job}}
@@ -163,10 +228,12 @@ defmodule Quantum.JobBroadcasterTest do
         TestScheduler.activate_job(broadcaster, active_job.name)
 
         refute_receive {:received, {:add, ^active_job}}
+
+        refute_receive {:update_job_state, {TestScheduler, _, _}, _}
       end)
     end
 
-    @tag jobs: :inactive
+    @tag jobs: :inactive, listen_storage: true
     test "inactive => inactive", %{broadcaster: broadcaster, inactive_job: inactive_job} do
       inactive_job_name = inactive_job.name
 
@@ -174,9 +241,12 @@ defmodule Quantum.JobBroadcasterTest do
         TestScheduler.deactivate_job(broadcaster, inactive_job.name)
 
         refute_receive {:received, {:remove, ^inactive_job_name}}
+
+        refute_receive {:update_job_state, {TestScheduler, _, _}, _}
       end)
     end
 
+    @tag listen_storage: true
     test "missing", %{broadcaster: broadcaster} do
       capture_log(fn ->
         TestScheduler.deactivate_job(broadcaster, make_ref())
@@ -184,12 +254,13 @@ defmodule Quantum.JobBroadcasterTest do
 
         refute_receive {:received, {:remove, _}}
         refute_receive {:received, {:add, _}}
+        refute_receive {:update_job_state, {TestScheduler, _, _}, _}
       end)
     end
   end
 
   describe "delete_all" do
-    @tag jobs: :both
+    @tag jobs: :both, listen_storage: true
     test "only active jobs", %{
       broadcaster: broadcaster,
       active_job: active_job,
@@ -203,6 +274,8 @@ defmodule Quantum.JobBroadcasterTest do
 
         refute_receive {:received, {:remove, ^inactive_job_name}}
         assert_receive {:received, {:remove, ^active_job_name}}
+
+        assert_receive {:purge, TestScheduler, _}
       end)
     end
   end
diff --git a/test/support/capture_log_extend.ex b/test/support/capture_log_extend.ex
new file mode 100644
index 0000000..b0becef
--- /dev/null
+++ b/test/support/capture_log_extend.ex
@@ -0,0 +1,20 @@
+defmodule Quantum.CaptureLogExtend do
+  @moduledoc false
+
+  import ExUnit.CaptureLog
+
+  def capture_log_with_return(fun) do
+    ref = make_ref()
+
+    logs =
+      capture_log(fn ->
+        return = fun.()
+        send(self(), {:return, ref, return})
+      end)
+
+    receive do
+      {:return, ^ref, return} ->
+        {return, logs}
+    end
+  end
+end
diff --git a/test/support/test_storage.ex b/test/support/test_storage.ex
new file mode 100644
index 0000000..a1a1a25
--- /dev/null
+++ b/test/support/test_storage.ex
@@ -0,0 +1,91 @@
+defmodule Quantum.Storage.Test do
+  @moduledoc """
+  Test implementation of a `Quantum.Storage.Adapter`.
+  """
+
+  @behaviour Quantum.Storage.Adapter
+
+  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})
+
+  def delete_job(scheduler_module, job_name),
+    do: send_and_wait(:delete_job, {scheduler_module, job_name})
+
+  def update_job_state(scheduler_module, job_name, state),
+    do: send_and_wait(:update_job_state, {scheduler_module, job_name, state})
+
+  def last_execution_date(scheduler_module),
+    do: send_and_wait(:last_execution_date, scheduler_module, :unknown)
+
+  def update_last_execution_date(scheduler_module, last_execution_date),
+    do: send_and_wait(:update_last_execution_date, {scheduler_module, last_execution_date})
+
+  def purge(scheduler_module), do: send_and_wait(:purge, scheduler_module)
+
+  @doc false
+  # Used for Small Test Storages
+  defmacro __using__(_) do
+    quote do
+      @behaviour Quantum.Storage.Adapter
+
+      alias Quantum.Storage.Test
+
+      def jobs(scheduler_module), do: Test.send_and_wait(:jobs, scheduler_module, :not_applicable)
+
+      def add_job(scheduler_module, job),
+        do: Test.send_and_wait(:add_job, {scheduler_module, job})
+
+      def delete_job(scheduler_module, job_name),
+        do: Test.send_and_wait(:delete_job, {scheduler_module, job_name})
+
+      def update_job_state(scheduler_module, job_name, state),
+        do: Test.send_and_wait(:update_job_state, {scheduler_module, job_name, state})
+
+      def last_execution_date(scheduler_module),
+        do: Test.send_and_wait(:last_execution_date, scheduler_module, :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})
+
+      def purge(scheduler_module), do: Test.send_and_wait(:purge, scheduler_module)
+
+      defoverridable Quantum.Storage.Adapter
+    end
+  end
+
+  def send_and_wait(fun, args, default \\ :ok) do
+    test_pid = find_test_pid(self())
+
+    if !is_nil(test_pid) do
+      ref = make_ref()
+
+      send(test_pid, {fun, args, {self(), ref}})
+    end
+
+    default
+  end
+
+  defp find_test_pid(pid) do
+    pid
+    |> Process.info()
+    |> case do
+      nil -> []
+      other -> other
+    end
+    |> Keyword.get(:dictionary, [])
+    |> Enum.into(%{})
+    |> case do
+      %{test_pid: pid} ->
+        pid
+
+      %{"$ancestors": ancestors} ->
+        Enum.find_value(ancestors, fn ancestor_pid ->
+          find_test_pid(ancestor_pid)
+        end)
+
+      _ ->
+        nil
+    end
+  end
+end