Solution: Storage Adapter

Author: maennchen

Diff for: 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,
@@ -45,13 +46,15 @@ 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)
     |> Keyword.put_new(:execution_broadcaster, execution_broadcaster)
     |> 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 for: 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,50 @@ defmodule Quantum.ExecutionBroadcaster do
     * `job_broadcaster` - The name of the stage to listen to
 
   """
-  @spec start_link(GenServer.server(), GenServer.server()) :: GenServer.on_start()
-  def start_link(name, job_broadcaster) do
+  @spec start_link(GenServer.server(), GenServer.server(), Adapter, Scheduler) ::
+          GenServer.on_start()
+  def start_link(name, job_broadcaster, storage, scheduler) do
     __MODULE__
-    |> GenStage.start_link(job_broadcaster, name: name)
+    |> GenStage.start_link({job_broadcaster, storage, scheduler}, name: name)
     |> Util.start_or_link()
   end
 
   @doc false
-  @spec child_spec({GenServer.server(), GenServer.server()}) :: Supervisor.child_spec()
-  def child_spec({name, job_broadcaster}) do
-    %{super([]) | start: {__MODULE__, :start_link, [name, job_broadcaster]}}
+  @spec child_spec({GenServer.server(), GenServer.server(), Adapter, Scheduler}) ::
+          Supervisor.child_spec()
+  def child_spec({name, job_broadcaster, storage, scheduler}) do
+    %{super([]) | start: {__MODULE__, :start_link, [name, job_broadcaster, storage, scheduler]}}
   end
 
   @doc false
-  def init(job_broadcaster) do
-    state = %{jobs: [], time: NaiveDateTime.utc_now(), timer: nil}
+  def init({job_broadcaster, storage, scheduler}) do
+    last_execution_date =
+      case storage.last_execution_date(scheduler) do
+        %NaiveDateTime{} = date ->
+          Logger.debug(fn ->
+            "[#{inspect(Node.self())}][#{__MODULE__}] Using last known execution time #{
+              NaiveDateTime.to_iso8601(date)
+            }"
+          end)
+
+          date
+
+        :unknown ->
+          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
+    }
+
     {:producer_consumer, state, subscribe_to: [job_broadcaster]}
   end
 
@@ -68,7 +99,16 @@ defmodule Quantum.ExecutionBroadcaster do
     {:noreply, reboot_add_events, state}
   end
 
-  def handle_info(:execute, %{jobs: [{time_to_execute, jobs_to_execute} | tail]} = state) do
+  def handle_info(
+        :execute,
+        %{
+          jobs: [{time_to_execute, jobs_to_execute} | tail],
+          storage: storage,
+          scheduler: scheduler
+        } = state
+      ) do
+    :ok = storage.update_last_execution_date(scheduler, time_to_execute)
+
     state =
       state
       |> Map.put(:timer, nil)
@@ -119,6 +159,8 @@ defmodule Quantum.ExecutionBroadcaster do
       end)
 
     %{state | jobs: jobs}
+    |> sort_state
+    |> reset_timer
   end
 
   defp add_job_to_state(
@@ -155,7 +197,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 for: 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 for: 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,45 @@ defmodule Quantum.JobBroadcaster do
    * `jobs` - Array of `Quantum.Job`
 
   """
-  @spec start_link(GenServer.server(), [Job.t()]) :: GenServer.on_start()
-  def start_link(name, jobs) do
+  @spec start_link(GenServer.server(), [Job.t()], Adapter, Scheduler) :: GenServer.on_start()
+  def start_link(name, jobs, storage, scheduler) do
     __MODULE__
-    |> GenStage.start_link(jobs, name: name)
+    |> GenStage.start_link({jobs, storage, scheduler}, name: name)
     |> Util.start_or_link()
   end
 
   @doc false
-  @spec child_spec({GenServer.server(), [Job.t()]}) :: Supervisor.child_spec()
-  def child_spec({name, jobs}) do
-    %{super([]) | start: {__MODULE__, :start_link, [name, jobs]}}
+  @spec child_spec({GenServer.server(), [Job.t()], Adapter, Scheduler}) :: Supervisor.child_spec()
+  def child_spec({name, jobs, storage, scheduler}) do
+    %{super([]) | start: {__MODULE__, :start_link, [name, jobs, storage, scheduler]}}
   end
 
   @doc false
-  def init(jobs) do
+  def init({jobs, storage, scheduler}) do
+    effective_jobs =
+      scheduler
+      |> storage.jobs()
+      |> case do
+        :not_applicable ->
+          Logger.debug(fn ->
+            "[#{inspect(Node.self())}][#{__MODULE__}] Loading Initial Jobs from Config"
+          end)
+
+          jobs
+
+        storage_jobs when is_list(storage_jobs) ->
+          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
     }
 
     {:producer, state}
@@ -47,40 +69,57 @@ defmodule Quantum.JobBroadcaster do
     {:noreply, to_send, %{state | buffer: remaining}}
   end
 
-  def handle_cast({:add, %Job{state: :active, name: job_name} = job}, %{jobs: jobs} = state) do
+  def handle_cast(
+        {:add, %Job{state: :active, name: job_name} = job},
+        %{jobs: jobs, storage: storage, scheduler: scheduler} = state
+      ) do
     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} = state) do
+  def handle_cast(
+        {:add, %Job{state: :inactive, name: job_name} = job},
+        %{jobs: jobs, storage: storage, scheduler: scheduler} = state
+      ) do
     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} = state) do
+  def handle_cast({:delete, name}, %{jobs: jobs, storage: storage, scheduler: scheduler} = state) do
     Logger.debug(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Deleting job #{inspect(name)}"
     end)
 
     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 ->
         {:noreply, [], state}
     end
   end
 
-  def handle_cast({:change_state, name, new_state}, %{jobs: jobs} = state) do
+  def handle_cast(
+        {:change_state, name, new_state},
+        %{jobs: jobs, storage: storage, scheduler: scheduler} = state
+      ) do
     Logger.debug(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Change job state #{inspect(name)}"
     end)
@@ -95,6 +134,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}}
@@ -105,13 +146,15 @@ defmodule Quantum.JobBroadcaster do
     end
   end
 
-  def handle_cast(:delete_all, %{jobs: jobs} = state) do
+  def handle_cast(:delete_all, %{jobs: jobs, storage: storage, scheduler: scheduler} = state) do
     Logger.debug(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Deleting all jobs"
     end)
 
     messages = for {name, %Job{state: :active}} <- jobs, do: {:remove, name}
 
+    :ok = storage.purge(scheduler)
+
     {:noreply, messages, %{state | jobs: %{}}}
   end
 

Diff for: lib/quantum/storage/adapter.ex

@@ -0,0 +1,61 @@
+defmodule Quantum.Storage.Adapter do
+  @moduledoc """
+  Bahaviour to be implemented by all Storage Adapters.
+  """
+
+  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 for: 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 for: lib/quantum/task_stages_supervisor.ex

@@ -32,14 +32,18 @@ defmodule Quantum.TaskStagesSupervisor do
           Quantum.JobBroadcaster,
           {
             Keyword.fetch!(opts, :job_broadcaster),
-            Keyword.fetch!(opts, :jobs)
+            Keyword.fetch!(opts, :jobs),
+            Keyword.fetch!(opts, :storage),
+            Keyword.fetch!(opts, :quantum)
           }
         },
         {
           Quantum.ExecutionBroadcaster,
           {
             Keyword.fetch!(opts, :execution_broadcaster),
-            Keyword.fetch!(opts, :job_broadcaster)
+            Keyword.fetch!(opts, :job_broadcaster),
+            Keyword.fetch!(opts, :storage),
+            Keyword.fetch!(opts, :quantum)
           }
         },
         {