Use crontab DateTime (TZ) Support

Author: maennchen

lib/quantum/clock_broadcaster.ex

@@ -44,11 +44,11 @@ defmodule Quantum.ClockBroadcaster do
       |> storage.last_execution_date()
       |> case do
         :unknown -> start_time
-        date -> date
+        date -> DateTime.from_naive!(date, "Etc/UTC")
       end
-      |> NaiveDateTime.truncate(:second)
+      |> DateTime.truncate(:second)
       # Roll back one second since handle_tick will start at `now + 1`.
-      |> NaiveDateTime.add(-1, :second)
+      |> DateTime.add(-1, :second)
 
     :timer.send_interval(1000, :tick)
 
@@ -80,16 +80,16 @@ defmodule Quantum.ClockBroadcaster do
 
   defp handle_tick(%State{remaining_demand: remaining_demand, time: time} = state)
        when remaining_demand > 0 do
-    now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
+    now = DateTime.truncate(DateTime.utc_now(), :second)
 
     {events, new_time} =
       Enum.reduce_while(
         1..remaining_demand,
         {[], time},
         fn _, {list, time} = acc ->
-          new_time = NaiveDateTime.add(time, 1, :second)
+          new_time = DateTime.add(time, 1, :second)
 
-          case NaiveDateTime.compare(new_time, now) do
+          case DateTime.compare(new_time, now) do
             :lt ->
               {:cont, {[%Event{time: new_time, catch_up: true} | list], new_time}}
 

lib/quantum/clock_broadcaster/event.ex

@@ -4,7 +4,7 @@ defmodule Quantum.ClockBroadcaster.Event do
   # Clock Event
 
   @type t :: %__MODULE__{
-          time: NaiveDateTime.t(),
+          time: DateTime.t(),
           catch_up: boolean()
         }
 

lib/quantum/clock_broadcaster/init_opts.ex

@@ -6,7 +6,7 @@ defmodule Quantum.ClockBroadcaster.InitOpts do
   alias Quantum.{Scheduler, Storage}
 
   @type t :: %__MODULE__{
-          start_time: NaiveDateTime.t(),
+          start_time: DateTime.t(),
           storage: Storage,
           scheduler: Scheduler,
           debug_logging: boolean()

lib/quantum/clock_broadcaster/start_opts.ex

@@ -7,7 +7,7 @@ defmodule Quantum.ClockBroadcaster.StartOpts do
 
   @type t :: %__MODULE__{
           name: GenServer.server(),
-          start_time: NaiveDateTime.t(),
+          start_time: DateTime.t(),
           storage: Storage,
           scheduler: Scheduler,
           debug_logging: boolean()

lib/quantum/clock_broadcaster/state.ex

@@ -5,7 +5,7 @@ defmodule Quantum.ClockBroadcaster.State do
 
   @type t :: %__MODULE__{
           debug_logging: boolean(),
-          time: NaiveDateTime.t(),
+          time: DateTime.t(),
           remaining_demand: non_neg_integer
         }
 

lib/quantum/date_library.ex

@@ -12,12 +12,6 @@ defmodule Quantum.ExecutionBroadcaster do
 
   alias Quantum.ClockBroadcaster.Event, as: ClockEvent
 
-  alias Quantum.{
-    DateLibrary,
-    DateLibrary.InvalidDateTimeForTimezoneError,
-    DateLibrary.InvalidTimezoneError
-  }
-
   alias Quantum.ExecutionBroadcaster.Event, as: ExecuteEvent
   alias Quantum.ExecutionBroadcaster.InitOpts
   alias Quantum.ExecutionBroadcaster.State
@@ -202,7 +196,7 @@ defmodule Quantum.ExecutionBroadcaster do
          } = state,
          time
        ) do
-    case NaiveDateTime.compare(time, time_to_execute) do
+    case DateTime.compare(time, time_to_execute) do
       :gt ->
         raise "Jobs were skipped"
 
@@ -226,7 +220,7 @@ defmodule Quantum.ExecutionBroadcaster do
           jobs
           |> Enum.reduce(
             %{state | execution_timeline: tail},
-            &add_job_to_state(&1, &2, NaiveDateTime.add(time, 1, :second))
+            &add_job_to_state(&1, &2, DateTime.add(time, 1, :second))
           )
           |> sort_state
           |> execute_events_to_fire(time)
@@ -240,11 +234,17 @@ defmodule Quantum.ExecutionBroadcaster do
          state,
          time
        ) do
-    job
-    |> get_next_execution_time(time)
-    |> case do
-      {:ok, date} ->
-        add_to_state(state, time, date, job)
+    with {:ok, execution_date} <- get_next_execution_time(job, time) do
+      add_to_state(state, time, execution_date, job)
+    else
+      {:error, :time_zone_not_found} ->
+        Logger.error(
+          "Invalid Timezone #{inspect(timezone)} provided for job #{inspect(name)}.",
+          job: job,
+          error: :time_zone_not_found
+        )
+
+        state
 
       {:error, _} ->
         Logger.warning(fn ->
@@ -256,54 +256,32 @@ defmodule Quantum.ExecutionBroadcaster do
 
         state
     end
-  rescue
-    e in InvalidTimezoneError ->
-      Logger.error(
-        "Invalid Timezone #{inspect(timezone)} provided for job #{inspect(name)}.",
-        job: job,
-        error: e
-      )
-
-      state
   end
 
   defp get_next_execution_time(
-         %Job{schedule: schedule, timezone: timezone, name: name} = job,
+         %Job{schedule: schedule, timezone: :utc},
          time
        ) do
-    schedule
-    |> CrontabScheduler.get_next_run_date(DateLibrary.to_tz!(time, timezone))
-    |> case do
-      {:ok, date} ->
-        {:ok, DateLibrary.to_utc!(date, timezone)}
+    CrontabScheduler.get_next_run_date(schedule, time)
+  end
 
-      {:error, _} = error ->
-        error
+  defp get_next_execution_time(
+         %Job{schedule: schedule, timezone: timezone},
+         time
+       ) do
+    with {:ok, localized_time} <- DateTime.shift_zone(time, timezone),
+         {:ok, localized_execution_time} <-
+           CrontabScheduler.get_next_run_date(schedule, localized_time) do
+      DateTime.shift_zone(localized_execution_time, "Etc/UTC")
     end
-  rescue
-    _ in InvalidDateTimeForTimezoneError ->
-      next_time = NaiveDateTime.add(time, 60, :second)
-
-      Logger.warning(fn ->
-        """
-        Next execution time for job #{inspect(name)} is not a valid time.
-        Retrying with #{inspect(next_time)}
-        """
-      end)
-
-      get_next_execution_time(job, next_time)
   end
 
   defp sort_state(%State{execution_timeline: execution_timeline} = state) do
-    %{
-      state
-      | execution_timeline:
-          Enum.sort_by(execution_timeline, fn {date, _} -> NaiveDateTime.to_erl(date) end)
-    }
+    %{state | execution_timeline: Enum.sort_by(execution_timeline, &elem(&1, 0), DateTime)}
   end
 
   defp add_to_state(%State{execution_timeline: execution_timeline} = state, time, date, job) do
-    unless NaiveDateTime.compare(time, date) in [:lt, :eq] do
+    unless DateTime.compare(time, date) in [:lt, :eq] do
       raise Quantum.ExecutionBroadcaster.JobInPastError
     end
 

lib/quantum/execution_broadcaster.ex

@@ -8,7 +8,7 @@ defmodule Quantum.ExecutionBroadcaster.State do
 
   @type t :: %__MODULE__{
           uninitialized_jobs: [Job.t()],
-          execution_timeline: [{NaiveDateTime.t(), [Job.t()]}],
+          execution_timeline: [{DateTime.t(), [Job.t()]}],
           storage: StorageAdapter,
           storage_pid: StorageAdapter.storage_pid(),
           scheduler: Quantum,

lib/quantum/execution_broadcaster/state.ex

@@ -62,14 +62,14 @@ 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(storage_pid :: storage_pid) :: :unknown | NaiveDateTime.t()
+  @callback last_execution_date(storage_pid :: storage_pid) :: :unknown | DateTime.t()
 
   @doc """
   Update last execution time to given date.
   """
   @callback update_last_execution_date(
               storage_pid :: storage_pid,
-              last_execution_date :: NaiveDateTime.t()
+              last_execution_date :: DateTime.t()
             ) :: :ok
 
   @doc """

lib/quantum/storage.ex

@@ -46,7 +46,7 @@ defmodule Quantum.Supervisor do
         opts
         |> Map.take([:debug_logging, :storage, :scheduler])
         |> Map.put(:name, clock_broadcaster_name)
-        |> Map.put(:start_time, NaiveDateTime.utc_now())
+        |> Map.put(:start_time, DateTime.utc_now())
       )
 
     job_broadcaster_opts =