Repository: quantum-elixir/quantum-core
Pull request #364 (2018-08-15 08:48:53)

Solution: Fix Worker Config in Global & Cleanup Worker Options
Do not merge until testing is done together with @zacksiri

coveralls: ## Pull Request Test Coverage Report for [Build 807](https://coveralls.io/builds/18493024)

* **3** of **3**   **(100.0%)**  changed or added relevant lines in **3** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **83.623%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/18493024/badge)](https://coveralls.io/builds/18493024) |
| :-- | --: |
| Change from base [Build 804](https://coveralls.io/builds/18458930): |  0.0% |
| Covered Lines: | 337 |
| Relevant Lines: | 403 |

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

maennchen: @c-rack Ready for Review

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 288ecc8..f949d0a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,9 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 Diff for [unreleased]
 
+### Fixed
+- Global Clustering Worker Start
+
 ## 2.3.1 - 2018-08-13
 
 ### Fixed
diff --git a/dialyzer.ignore-warnings b/dialyzer.ignore-warnings
index 874bc1e..a295591 100644
--- a/dialyzer.ignore-warnings
+++ b/dialyzer.ignore-warnings
@@ -1,24 +1,19 @@
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Atom':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.BitString':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Float':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Function':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Integer':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.List':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Map':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.PID':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Port':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Reference':'__impl__'/1
-Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Tuple':'__impl__'/1
-Overloaded contract for 'Elixir.Quantum.DateLibrary':'to_utc!'/2 has overlapping domains; such contracts are currently unsupported and are simply ignored
-lib/quantum/execution_broadcaster.ex:201: The inferred return type of handle_call/3 ({'reply',{'resume',{_,_}},#{'jobs':=_, 'last_execution_date':=_, _=>_}}) has nothing in common with {'noreply',[any()],_} | {'stop',_,_} | {'noreply',[any()],_,'hibernate'} | {'reply',_,[any()],_} | {'stop',_,_,_} | {'reply',_,[any()],_,'hibernate'}, which is the expected return type for the callback of the 'Elixir.GenStage' behaviour
-lib/quantum/execution_broadcaster.ex:213: The inferred return type of handle_cast/2 ({'noreply',_}) has nothing in common with {'noreply',[any()],_} | {'stop',_,_} | {'noreply',[any()],_,'hibernate'}, which is the expected return type for the callback of the 'Elixir.GenStage' behaviour
-lib/quantum/execution_broadcaster.ex:266: The pattern {'ok', Vdate@1} can never match the type {'error',<<_:368>>}
-lib/quantum/execution_broadcaster.ex:295: The pattern {'ok', Vdate@1} can never match the type {'error',<<_:368>>}
-lib/quantum/execution_broadcaster.ex:319: Function add_to_state/3 will never be called
-lib/quantum/execution_broadcaster.ex:327: Function add_job_at_date/3 will never be called
-lib/quantum/execution_broadcaster.ex:334: Function find_date_and_put_job/3 will never be called
-lib/quantum/executor.ex:23: Overloaded contract for 'Elixir.Quantum.Executor':start_link/2 has overlapping domains; such contracts are currently unsupported and are simply ignored
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Atom':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.BitString':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Float':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Function':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Integer':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.List':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Map':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.PID':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Port':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Reference':'__impl__'/1
+:0: Unknown function 'Elixir.Quantum.RunStrategy.NodeList.Tuple':'__impl__'/1
+lib/quantum/date_library.ex:38: Overloaded contract for 'Elixir.Quantum.DateLibrary':'to_utc!'/2 has overlapping domains; such contracts are currently unsupported and are simply ignored
+lib/quantum/execution_broadcaster.ex:278: The pattern {'ok', Vdate@1} can never match the type {'error',<<_:368>>}
+lib/quantum/execution_broadcaster.ex:307: The pattern {'ok', Vdate@1} can never match the type {'error',<<_:368>>}
+lib/quantum/execution_broadcaster.ex:331: Function add_to_state/3 will never be called
+lib/quantum/execution_broadcaster.ex:339: Function add_job_at_date/3 will never be called
+lib/quantum/execution_broadcaster.ex:346: Function find_date_and_put_job/3 will never be called
 lib/quantum/util.ex:22: Function 'gen_stage_v12?'/0 has no local return
 lib/quantum/util.ex:31: The call 'Elixir.Version':'match?'(binary(),<<_:64>>) will never return since it differs in the 1st argument from the success typing arguments: (#{'__struct__':='Elixir.Version', 'build':='nil' | binary(), 'major':=binary() | non_neg_integer(), 'minor':='nil' | non_neg_integer(), 'patch':='nil' | non_neg_integer(), 'pre':=[binary() | non_neg_integer()]},binary() | #{'__struct__':='Elixir.Version.Requirement', 'compiled':=boolean(), 'matchspec':=[{atom() | tuple(),[any()],[any()]}] | ets:comp_match_spec(), 'source':=_})
-lib/quantum/execution_broadcaster.ex:201: The inferred return type of handle_call/3 ({'reply',{'resume',{_,_}},#{'jobs':=_, 'last_execution_date':=_, _=>_}}) has nothing in common with {'noreply',[any()],_} | {'stop',_,_} | {'noreply',[any()],_,'hibernate'} | {'reply',_,[any()],_} | {'stop',_,_,_} | {'reply',_,[any()],_,'hibernate'}, which is the expected return type for the callback of 'Elixir.GenStage' behaviour
-lib/quantum/execution_broadcaster.ex:213: The inferred return type of handle_cast/2 ({'noreply',_}) has nothing in common with {'noreply',[any()],_} | {'stop',_,_} | {'noreply',[any()],_,'hibernate'}, which is the expected return type for the callback of 'Elixir.GenStage' behaviour
diff --git a/lib/quantum.ex b/lib/quantum.ex
index 66dd793..74d1971 100644
--- a/lib/quantum.ex
+++ b/lib/quantum.ex
@@ -28,38 +28,63 @@ defmodule Quantum do
       |> Keyword.merge(custom)
       |> Keyword.merge(otp_app: otp_app, quantum: quantum)
 
-    # Default Job Broadcaster Name
     global = Keyword.fetch!(config, :global)
-    job_broadcaster = cluster_worker_config(Module.concat(quantum, JobBroadcaster), global)
 
-    execution_broadcaster =
-      cluster_worker_config(Module.concat(quantum, ExecutionBroadcaster), global)
+    job_broadcaster_name = Module.concat(quantum, JobBroadcaster)
 
-    executor_supervisor = Module.concat(quantum, ExecutorSupervisor)
+    job_broadcaster_reference =
+      if global,
+        do: {:via, :swarm, job_broadcaster_name},
+        else: job_broadcaster_name
 
-    task_registry = cluster_worker_config(Module.concat(quantum, TaskRegistry), global)
+    execution_broadcaster_name = Module.concat(quantum, ExecutionBroadcaster)
 
-    # Default Task Supervisor Name
-    cluster_task_supervisor_registry = Module.concat(quantum, ClusterTaskSupervisorRegistry)
-    task_supervisor = Module.concat(quantum, Task.Supervisor)
+    execution_broadcaster_reference =
+      if global,
+        do: {:via, :swarm, execution_broadcaster_name},
+        else: execution_broadcaster_name
+
+    executor_supervisor_name = Module.concat(quantum, ExecutorSupervisor)
+    executor_supervisor_reference = executor_supervisor_name
+
+    task_registry_name = Module.concat(quantum, TaskRegistry)
+
+    task_registry_reference =
+      if global,
+        do: {:via, :swarm, task_registry_name},
+        else: task_registry_name
+
+    cluster_task_supervisor_registry_name = Module.concat(quantum, ClusterTaskSupervisorRegistry)
+    cluster_task_supervisor_registry_reference = cluster_task_supervisor_registry_name
+
+    task_supervisor_name = Module.concat(quantum, Task.Supervisor)
+    task_supervisor_reference = task_supervisor_name
 
     config
     |> Keyword.put_new(:quantum, quantum)
+    |> Keyword.put_new(:scheduler, quantum)
     |> update_in([:schedule], &Normalizer.normalize_schedule/1)
-    |> 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(:cluster_task_supervisor_registry, cluster_task_supervisor_registry)
+    |> Keyword.put_new(:job_broadcaster_name, job_broadcaster_name)
+    |> Keyword.put_new(:job_broadcaster_reference, job_broadcaster_reference)
+    |> Keyword.put_new(:execution_broadcaster_name, execution_broadcaster_name)
+    |> Keyword.put_new(:execution_broadcaster_reference, execution_broadcaster_reference)
+    |> Keyword.put_new(:executor_supervisor_name, executor_supervisor_name)
+    |> Keyword.put_new(:executor_supervisor_reference, executor_supervisor_reference)
+    |> Keyword.put_new(:task_registry_name, task_registry_name)
+    |> Keyword.put_new(:task_registry_reference, task_registry_reference)
+    |> Keyword.put_new(:task_supervisor_name, task_supervisor_name)
+    |> Keyword.put_new(:task_supervisor_reference, task_supervisor_reference)
+    |> Keyword.put_new(
+      :cluster_task_supervisor_registry_name,
+      cluster_task_supervisor_registry_name
+    )
+    |> Keyword.put_new(
+      :cluster_task_supervisor_registry_reference,
+      cluster_task_supervisor_registry_reference
+    )
     |> Keyword.put_new(:storage, Noop)
   end
 
-  defp cluster_worker_config(module, false), do: [name: module, restart: :permanent]
-
-  defp cluster_worker_config(module, true),
-    do: [name: {:via, :swarm, module}, restart: :temporary]
-
   @doc """
   Retrieves the comprehensive runtime configuration.
   """
diff --git a/lib/quantum/cluster_task_supervisor_registry.ex b/lib/quantum/cluster_task_supervisor_registry.ex
index 2f27d29..f77fceb 100644
--- a/lib/quantum/cluster_task_supervisor_registry.ex
+++ b/lib/quantum/cluster_task_supervisor_registry.ex
@@ -4,23 +4,26 @@ defmodule Quantum.ClusterTaskSupervisorRegistry do
 
   use GenServer
 
-  @doc false
-  def start_link(opts) do
-    name = Keyword.get(opts, :name, __MODULE__)
+  alias __MODULE__.{InitOpts, StartOpts, State}
 
+  @doc false
+  @spec start_link(StartOpts.t()) :: GenServer.on_start()
+  def start_link(%StartOpts{name: name} = opts) do
     GenServer.start_link(
       __MODULE__,
-      {
-        Keyword.fetch!(opts, :task_supervisor),
-        Keyword.get(opts, :group_name, Module.concat(name, Group))
-      },
+      struct!(
+        InitOpts,
+        opts
+        |> Map.take([:task_supervisor_reference, :group_name])
+        |> Map.put_new(:group_name, Module.concat(name, Group))
+      ),
       name: name
     )
   end
 
   @doc false
   @impl true
-  def init({task_supervisor, group_name}) do
+  def init(%InitOpts{task_supervisor_reference: task_supervisor, group_name: group_name}) do
     task_supervisor_pid = GenServer.whereis(task_supervisor)
 
     monitor_ref = Process.monitor(task_supervisor_pid)
@@ -35,12 +38,16 @@ defmodule Quantum.ClusterTaskSupervisorRegistry do
     :ok = Swarm.join(group_name, task_supervisor_pid)
 
     {:ok,
-     %{group_name: group_name, task_supervisor_pid: task_supervisor_pid, monitor_ref: monitor_ref}}
+     %State{
+       group_name: group_name,
+       task_supervisor_pid: task_supervisor_pid,
+       monitor_ref: monitor_ref
+     }}
   end
 
   @doc false
   @impl true
-  def handle_call(:pids, _from, %{group_name: group_name} = state) do
+  def handle_call(:pids, _from, %State{group_name: group_name} = state) do
     {:reply, Swarm.members(group_name), state}
   end
 
@@ -48,7 +55,7 @@ defmodule Quantum.ClusterTaskSupervisorRegistry do
   @impl true
   def handle_info(
         {:DOWN, monitor_ref, :process, task_supervisor_pid, _reason},
-        %{
+        %State{
           group_name: group_name,
           task_supervisor_pid: task_supervisor_pid,
           monitor_ref: monitor_ref
diff --git a/lib/quantum/cluster_task_supervisor_registry/init_opts.ex b/lib/quantum/cluster_task_supervisor_registry/init_opts.ex
new file mode 100644
index 0000000..f87391b
--- /dev/null
+++ b/lib/quantum/cluster_task_supervisor_registry/init_opts.ex
@@ -0,0 +1,13 @@
+defmodule Quantum.ClusterTaskSupervisorRegistry.InitOpts do
+  @moduledoc false
+
+  # Init Options for Quantum.ClusterTaskSupervisorRegistry
+
+  @type t :: %__MODULE__{
+          task_supervisor_reference: GenServer.server(),
+          group_name: atom()
+        }
+
+  @enforce_keys [:task_supervisor_reference, :group_name]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/cluster_task_supervisor_registry/start_opts.ex b/lib/quantum/cluster_task_supervisor_registry/start_opts.ex
new file mode 100644
index 0000000..ddefe67
--- /dev/null
+++ b/lib/quantum/cluster_task_supervisor_registry/start_opts.ex
@@ -0,0 +1,14 @@
+defmodule Quantum.ClusterTaskSupervisorRegistry.StartOpts do
+  @moduledoc false
+
+  # Start Options for Quantum.ClusterTaskSupervisorRegistry
+
+  @type t :: %__MODULE__{
+          name: GenServer.server(),
+          task_supervisor_reference: GenServer.server(),
+          group_name: atom() | nil
+        }
+
+  @enforce_keys [:name, :task_supervisor_reference]
+  defstruct @enforce_keys ++ [:group_name]
+end
diff --git a/lib/quantum/cluster_task_supervisor_registry/state.ex b/lib/quantum/cluster_task_supervisor_registry/state.ex
new file mode 100644
index 0000000..4ac7744
--- /dev/null
+++ b/lib/quantum/cluster_task_supervisor_registry/state.ex
@@ -0,0 +1,14 @@
+defmodule Quantum.ClusterTaskSupervisorRegistry.State do
+  @moduledoc false
+
+  # State of Quantum.ClusterTaskSupervisorRegistry
+
+  @type t :: %__MODULE__{
+          group_name: atom(),
+          task_supervisor_pid: GenServer.server(),
+          monitor_ref: reference
+        }
+
+  @enforce_keys [:group_name, :task_supervisor_pid, :monitor_ref]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/execution_broadcaster.ex b/lib/quantum/execution_broadcaster.ex
index 46d8b0d..30f785f 100644
--- a/lib/quantum/execution_broadcaster.ex
+++ b/lib/quantum/execution_broadcaster.ex
@@ -19,6 +19,10 @@ defmodule Quantum.ExecutionBroadcaster do
   alias Crontab.CronExpression
   alias Crontab.Scheduler, as: CrontabScheduler
 
+  alias __MODULE__.{InitOpts, StartOpts, State}
+
+  @type event :: {:add, Job.t()} | {:execute, Job.t()}
+
   defmodule JobInPastError do
     defexception message:
                    "The job was scheduled in the past. This must not happen to prevent infinite loops!"
@@ -26,47 +30,55 @@ defmodule Quantum.ExecutionBroadcaster do
 
   @doc """
   Start Stage
-
-  ### Arguments
-
-    * `name` - The name of the stage
-    * `job_broadcaster` - The name of the stage to listen to
-
   """
-  @spec start_link(GenServer.server(), GenServer.server(), Adapter, Scheduler, boolean()) ::
-          GenServer.on_start()
-  def start_link(name, job_broadcaster, storage, scheduler, debug_logging) do
+  @spec start_link(StartOpts.t()) :: GenServer.on_start()
+  def start_link(%StartOpts{name: name} = opts) do
     GenStage.start_link(
       __MODULE__,
-      {job_broadcaster, storage, scheduler, debug_logging},
+      struct!(
+        InitOpts,
+        Map.take(opts, [:job_broadcaster_reference, :storage, :scheduler, :debug_logging])
+      ),
       name: name
     )
   end
 
+  @spec start_link(GenServer.server(), GenServer.server(), Adapter, Scheduler, boolean()) ::
+          GenServer.on_start()
+  def start_link(name, job_broadcaster, storage, scheduler, debug_logging),
+    do:
+      start_link(%StartOpts{
+        name: name,
+        job_broadcaster_reference: job_broadcaster,
+        storage: storage,
+        scheduler: scheduler,
+        debug_logging: debug_logging
+      })
+
   @doc false
-  @spec child_spec(
-          {Keyword.t() | GenServer.server(), GenServer.server(), Adapter, Scheduler, boolean()}
-        ) :: Supervisor.child_spec()
-  def child_spec({opts, job_broadcaster, storage, scheduler, debug_logging}) when is_list(opts) do
-    %{
-      super(opts)
-      | start:
-          {__MODULE__, :start_link,
-           [
-             Keyword.fetch!(opts, :name),
-             job_broadcaster,
-             storage,
-             scheduler,
-             debug_logging
-           ]}
-    }
-  end
+  @spec child_spec(StartOpts.t()) :: Supervisor.child_spec()
+  def child_spec(%StartOpts{} = opts), do: super(opts)
+
+  @spec child_spec({GenServer.server(), GenServer.server(), Adapter, Scheduler, boolean()}) ::
+          Supervisor.child_spec()
 
   def child_spec({name, job_broadcaster, storage, scheduler, debug_logging}),
-    do: child_spec({[name: name], job_broadcaster, storage, scheduler, debug_logging})
+    do:
+      child_spec(%StartOpts{
+        name: name,
+        job_broadcaster_reference: job_broadcaster,
+        storage: storage,
+        scheduler: scheduler,
+        debug_logging: debug_logging
+      })
 
   @doc false
-  def init({job_broadcaster, storage, scheduler, debug_logging}) do
+  def init(%InitOpts{
+        job_broadcaster_reference: job_broadcaster,
+        storage: storage,
+        scheduler: scheduler,
+        debug_logging: debug_logging
+      }) do
     last_execution_date =
       case storage.last_execution_date(scheduler) do
         %NaiveDateTime{} = date ->
@@ -88,19 +100,18 @@ defmodule Quantum.ExecutionBroadcaster do
           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]}
+    {:producer_consumer,
+     %State{
+       jobs: [],
+       time: last_execution_date,
+       timer: nil,
+       storage: storage,
+       scheduler: scheduler,
+       debug_logging: debug_logging
+     }, subscribe_to: [job_broadcaster]}
   end
 
-  def handle_events(events, _, %{debug_logging: debug_logging} = state) do
+  def handle_events(events, _, %State{debug_logging: debug_logging} = state) do
     reboot_add_events =
       events
       |> Enum.filter(&add_reboot_event?/1)
@@ -127,7 +138,7 @@ defmodule Quantum.ExecutionBroadcaster do
 
   def handle_info(
         :execute,
-        %{
+        %State{
           jobs: [{time_to_execute, jobs_to_execute} | tail],
           storage: storage,
           scheduler: scheduler,
@@ -136,11 +147,12 @@ defmodule Quantum.ExecutionBroadcaster do
       ) do
     :ok = storage.update_last_execution_date(scheduler, time_to_execute)
 
-    state =
+    state = %{
       state
-      |> Map.put(:timer, nil)
-      |> Map.put(:jobs, tail)
-      |> Map.put(:time, NaiveDateTime.add(time_to_execute, 1, :second))
+      | timer: nil,
+        jobs: tail,
+        time: NaiveDateTime.add(time_to_execute, 1, :second)
+    }
 
     state =
       jobs_to_execute
@@ -163,12 +175,12 @@ defmodule Quantum.ExecutionBroadcaster do
     {:noreply, Enum.map(jobs_to_execute, fn job -> {:execute, job} end), state}
   end
 
-  def handle_info({:swarm, :die}, %{timer: {timer, _}} = state) do
+  def handle_info({:swarm, :die}, %State{timer: {timer, _}} = state) do
     Process.cancel_timer(timer)
     {:stop, :shutdown, %{state | timer: nil}}
   end
 
-  defp handle_event({:add, %{name: job_name} = job}, %{debug_logging: debug_logging} = state) do
+  defp handle_event({:add, %{name: job_name} = job}, %State{debug_logging: debug_logging} = state) do
     debug_logging &&
       Logger.debug(fn ->
         "[#{inspect(Node.self())}][#{__MODULE__}] Adding job #{inspect(job_name)}"
@@ -177,7 +189,7 @@ defmodule Quantum.ExecutionBroadcaster do
     add_job_to_state(job, state)
   end
 
-  defp handle_event({:remove, name}, %{jobs: jobs, debug_logging: debug_logging} = state) do
+  defp handle_event({:remove, name}, %State{jobs: jobs, debug_logging: debug_logging} = state) do
     debug_logging &&
       Logger.debug(fn ->
         "[#{inspect(Node.self())}][#{__MODULE__}] Removing job #{inspect(name)}"
@@ -201,18 +213,18 @@ defmodule Quantum.ExecutionBroadcaster do
   def handle_call(
         {:swarm, :begin_handoff},
         _from,
-        %{jobs: jobs, last_execution_date: last_execution_date} = state
+        %State{jobs: jobs, time: last_execution_date} = state
       ) do
     Logger.info(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Handing of state to other cluster node"
     end)
 
-    {:reply, {:resume, {jobs, last_execution_date}}, state}
+    {:reply, {:resume, {jobs, last_execution_date}}, [], state}
   end
 
   def handle_cast(
         {:swarm, :end_handoff, {handoff_jobs, handoff_last_execution_time}},
-        %{last_execution_date: last_execution_date} = state
+        %State{time: last_execution_date} = state
       ) do
     Logger.info(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Incorperating state from other cluster node"
@@ -223,19 +235,19 @@ defmodule Quantum.ExecutionBroadcaster do
         do: handoff_last_execution_time,
         else: last_execution_date
 
-    intermediate_state = %{state | last_execution_date: earlier_last_execution_date}
+    intermediate_state = %{state | time: earlier_last_execution_date}
 
     new_state =
       Enum.reduce(handoff_jobs, intermediate_state, fn job, acc_state ->
         add_job_to_state(job, acc_state)
       end)
 
-    {:noreply, new_state}
+    {:noreply, [], new_state}
   end
 
   def handle_cast(
         {:swarm, :resolve_conflict, {handoff_jobs, handoff_last_execution_time}},
-        %{last_execution_date: last_execution_date} = state
+        %State{time: last_execution_date} = state
       ) do
     Logger.info(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Incorperating conflict state from other cluster node"
@@ -246,19 +258,19 @@ defmodule Quantum.ExecutionBroadcaster do
         do: handoff_last_execution_time,
         else: last_execution_date
 
-    intermediate_state = %{state | last_execution_date: earlier_last_execution_date}
+    intermediate_state = %{state | time: earlier_last_execution_date}
 
     new_state =
       Enum.reduce(handoff_jobs, intermediate_state, fn job, acc_state ->
         add_job_to_state(job, acc_state)
       end)
 
-    {:noreply, new_state}
+    {:noreply, [], new_state}
   end
 
   defp add_job_to_state(
          %Job{schedule: schedule, timezone: timezone, name: name} = job,
-         %{time: time} = state
+         %State{time: time} = state
        ) do
     job
     |> get_next_execution_time(time)
@@ -312,11 +324,11 @@ defmodule Quantum.ExecutionBroadcaster do
       get_next_execution_time(job, next_time)
   end
 
-  defp sort_state(%{jobs: jobs} = state) do
+  defp sort_state(%State{jobs: jobs} = state) do
     %{state | jobs: Enum.sort_by(jobs, fn {date, _} -> NaiveDateTime.to_erl(date) end)}
   end
 
-  defp add_to_state(%{jobs: jobs, time: time} = state, date, job) do
+  defp add_to_state(%State{jobs: jobs, time: time} = state, date, job) do
     unless NaiveDateTime.compare(time, date) in [:lt, :eq] do
       raise JobInPastError
     end
@@ -344,17 +356,17 @@ defmodule Quantum.ExecutionBroadcaster do
     {state, [head | new_rest]}
   end
 
-  defp reset_timer(%{timer: nil, jobs: []} = state) do
+  defp reset_timer(%State{timer: nil, jobs: []} = state) do
     state
   end
 
-  defp reset_timer(%{timer: {timer, _}, jobs: []} = state) do
+  defp reset_timer(%State{timer: {timer, _}, jobs: []} = state) do
     Process.cancel_timer(timer)
 
     Map.put(state, :timer, nil)
   end
 
-  defp reset_timer(%{timer: nil, jobs: jobs, debug_logging: debug_logging} = state) do
+  defp reset_timer(%State{timer: nil, jobs: jobs, debug_logging: debug_logging} = state) do
     run_date = next_run_date(jobs)
 
     timer =
@@ -387,10 +399,10 @@ defmodule Quantum.ExecutionBroadcaster do
           nil
       end
 
-    Map.put(state, :timer, {timer, run_date})
+    %{state | timer: {timer, run_date}}
   end
 
-  defp reset_timer(%{timer: {timer, old_date}, jobs: jobs} = state) do
+  defp reset_timer(%State{timer: {timer, old_date}, jobs: jobs} = state) do
     run_date = next_run_date(jobs)
 
     case NaiveDateTime.compare(run_date, old_date) do
diff --git a/lib/quantum/execution_broadcaster/init_opts.ex b/lib/quantum/execution_broadcaster/init_opts.ex
new file mode 100644
index 0000000..bd7c383
--- /dev/null
+++ b/lib/quantum/execution_broadcaster/init_opts.ex
@@ -0,0 +1,17 @@
+defmodule Quantum.ExecutionBroadcaster.InitOpts do
+  @moduledoc false
+
+  # Init Options for Quantum.ExecutionBroadcaster
+
+  alias Quantum.{Scheduler, Storage.Adapter}
+
+  @type t :: %__MODULE__{
+          job_broadcaster_reference: GenServer.server(),
+          storage: Adapter,
+          scheduler: Scheduler,
+          debug_logging: boolean
+        }
+
+  @enforce_keys [:job_broadcaster_reference, :storage, :scheduler, :debug_logging]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/execution_broadcaster/start_opts.ex b/lib/quantum/execution_broadcaster/start_opts.ex
new file mode 100644
index 0000000..6a10886
--- /dev/null
+++ b/lib/quantum/execution_broadcaster/start_opts.ex
@@ -0,0 +1,18 @@
+defmodule Quantum.ExecutionBroadcaster.StartOpts do
+  @moduledoc false
+
+  # Start Options for Quantum.ExecutionBroadcaster
+
+  alias Quantum.{Scheduler, Storage.Adapter}
+
+  @type t :: %__MODULE__{
+          name: GenServer.server(),
+          job_broadcaster_reference: GenServer.server(),
+          storage: Adapter,
+          scheduler: Scheduler,
+          debug_logging: boolean
+        }
+
+  @enforce_keys [:name, :job_broadcaster_reference, :storage, :scheduler, :debug_logging]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/execution_broadcaster/state.ex b/lib/quantum/execution_broadcaster/state.ex
new file mode 100644
index 0000000..ac6eed7
--- /dev/null
+++ b/lib/quantum/execution_broadcaster/state.ex
@@ -0,0 +1,19 @@
+defmodule Quantum.ExecutionBroadcaster.State do
+  @moduledoc false
+
+  # State of Quantum.ExecutionBroadcaster
+
+  alias Quantum.{Job, Scheduler, Storage}
+
+  @type t :: %__MODULE__{
+          jobs: [{NaiveDateTime.t(), [Job.t()]}],
+          time: NaiveDateTime.t(),
+          timer: {reference(), NaiveDateTime.t()} | nil,
+          storage: Storage,
+          scheduler: Scheduler,
+          debug_logging: boolean()
+        }
+
+  @enforce_keys [:jobs, :time, :timer, :storage, :scheduler, :debug_logging]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/executor.ex b/lib/quantum/executor.ex
index ae58072..498e111 100644
--- a/lib/quantum/executor.ex
+++ b/lib/quantum/executor.ex
@@ -1,7 +1,6 @@
 defmodule Quantum.Executor do
   @moduledoc """
   Task to actually execute a Task
-
   """
 
   use Task
@@ -10,6 +9,8 @@ defmodule Quantum.Executor do
 
   alias Quantum.{ClusterTaskSupervisorRegistry, Job, RunStrategy.NodeList, TaskRegistry}
 
+  alias __MODULE__.StartOpts
+
   @doc """
   Start the Task
 
@@ -22,47 +23,40 @@ defmodule Quantum.Executor do
   """
   @spec start_link({GenServer.server(), GenServer.server(), boolean()}, {:execute, Job.t()}) ::
           {:ok, pid}
-  def start_link({task_supervisor, task_registry, debug_logging}, {:execute, job}) do
-    Task.start_link(fn ->
-      execute(task_supervisor, task_registry, nil, debug_logging, job)
-    end)
-  end
+  def start_link({task_supervisor, task_registry, debug_logging}, event),
+    do:
+      start_link(
+        %StartOpts{
+          task_supervisor_reference: task_supervisor,
+          cluster_task_supervisor_registry_reference: nil,
+          task_registry_reference: task_registry,
+          debug_logging: debug_logging
+        },
+        event
+      )
 
-  @spec start_link({GenServer.server(), GenServer.server(), boolean()}, {:execute, Job.t()}) ::
-          {:ok, pid}
+  @spec start_link(StartOpts.t(), {:execute, Job.t()}) :: {:ok, pid}
   def start_link(
-        %{
-          task_supervisor: task_supervisor,
-          task_registry: task_registry,
-          debug_logging: debug_logging,
-          cluster_task_supervisor_registry: cluster_task_supervisor_registry
-        },
+        opts,
         {:execute, job}
       ) do
     Task.start_link(fn ->
-      execute(
-        task_supervisor,
-        task_registry,
-        cluster_task_supervisor_registry,
-        debug_logging,
-        job
-      )
+      execute(opts, job)
     end)
   end
 
   @spec execute(
-          GenServer.server(),
-          GenServer.server(),
-          GenServer.server() | nil,
-          boolean(),
+          StartOpts.t(),
           Job.t()
         ) :: :ok
   # Execute task on all given nodes without checking for overlap
   defp execute(
-         task_supervisor,
-         _task_registry,
-         cluster_task_supervisor_registry,
-         debug_logging,
+         %StartOpts{
+           task_supervisor_reference: task_supervisor,
+           task_registry_reference: _task_registry,
+           debug_logging: debug_logging,
+           cluster_task_supervisor_registry_reference: cluster_task_supervisor_registry
+         },
          %Job{overlap: true, run_strategy: run_strategy} = job
        ) do
     # Find Nodes to run on
@@ -77,10 +71,12 @@ defmodule Quantum.Executor do
 
   # Execute task on all given nodes with checking for overlap
   defp execute(
-         task_supervisor,
-         task_registry,
-         cluster_task_supervisor_registry,
-         debug_logging,
+         %StartOpts{
+           task_supervisor_reference: task_supervisor,
+           task_registry_reference: task_registry,
+           debug_logging: debug_logging,
+           cluster_task_supervisor_registry_reference: cluster_task_supervisor_registry
+         },
          %Job{overlap: false, run_strategy: run_strategy, name: job_name} = job
        ) do
     debug_logging &&
diff --git a/lib/quantum/executor/start_opts.ex b/lib/quantum/executor/start_opts.ex
new file mode 100644
index 0000000..b6c4751
--- /dev/null
+++ b/lib/quantum/executor/start_opts.ex
@@ -0,0 +1,20 @@
+defmodule Quantum.Executor.StartOpts do
+  @moduledoc false
+
+  # Start Options for Quantum.Executor
+
+  @type t :: %__MODULE__{
+          task_supervisor_reference: GenServer.server(),
+          task_registry_reference: GenServer.server(),
+          debug_logging: boolean,
+          cluster_task_supervisor_registry_reference: GenServer.server() | nil
+        }
+
+  @enforce_keys [
+    :task_supervisor_reference,
+    :task_registry_reference,
+    :debug_logging,
+    :cluster_task_supervisor_registry_reference
+  ]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/executor_supervisor.ex b/lib/quantum/executor_supervisor.ex
index 2495958..809ff8d 100644
--- a/lib/quantum/executor_supervisor.ex
+++ b/lib/quantum/executor_supervisor.ex
@@ -5,8 +5,11 @@ defmodule Quantum.ExecutorSupervisor do
 
   use ConsumerSupervisor
 
+  alias Quantum.Executor.StartOpts, as: ExecutorStartOpts
   alias Quantum.Util
 
+  alias __MODULE__.{InitOpts, StartOpts}
+
   @spec start_link(
           GenServer.server(),
           GenServer.server(),
@@ -14,39 +17,32 @@ defmodule Quantum.ExecutorSupervisor do
           GenServer.server(),
           boolean()
         ) :: GenServer.on_start()
-  def start_link(name, execution_broadcaster, task_supervisor, task_registry, debug_logging) do
-    ConsumerSupervisor.start_link(
-      __MODULE__,
-      %{
-        execution_broadcaster: execution_broadcaster,
-        task_supervisor: task_supervisor,
-        task_registry: task_registry,
-        cluster_task_supervisor_registry: nil,
+  def start_link(name, execution_broadcaster, task_supervisor, task_registry, debug_logging),
+    do:
+      start_link(%StartOpts{
+        name: name,
+        execution_broadcaster_reference: execution_broadcaster,
+        task_supervisor_reference: task_supervisor,
+        task_registry_reference: task_registry,
+        cluster_task_supervisor_registry_reference: nil,
         debug_logging: debug_logging
-      },
-      name: name
-    )
-  end
+      })
 
-  @spec start_link(%{
-          name: GenServer.server(),
-          execution_broadcaster: GenServer.server(),
-          task_supervisor: GenServer.server(),
-          task_registry: GenServer.server(),
-          cluster_task_supervisor_registry: nil | GenServer.server(),
-          debug_logging: boolean()
-        }) :: GenServer.on_start()
-  def start_link(opts) do
+  @spec start_link(StartOpts.t()) :: GenServer.on_start()
+  def start_link(%StartOpts{name: name} = opts) do
     ConsumerSupervisor.start_link(
       __MODULE__,
-      Map.take(opts, [
-        :execution_broadcaster,
-        :task_supervisor,
-        :task_registry,
-        :cluster_task_supervisor_registry,
-        :debug_logging
-      ]),
-      name: Map.fetch!(opts, :name)
+      struct!(
+        InitOpts,
+        Map.take(opts, [
+          :execution_broadcaster_reference,
+          :task_supervisor_reference,
+          :task_registry_reference,
+          :cluster_task_supervisor_registry_reference,
+          :debug_logging
+        ])
+      ),
+      name: name
     )
   end
 
@@ -54,17 +50,17 @@ defmodule Quantum.ExecutorSupervisor do
   # TODO: Remove when gen_stage:0.12 support is dropped
   if Util.gen_stage_v12?() do
     def init(
-          %{
+          %InitOpts{
             execution_broadcaster: execution_broadcaster
           } = opts
         ) do
       ConsumerSupervisor.init(
         {Quantum.Executor,
          Map.take(opts, [
-           :task_supervisor,
-           :task_registry,
+           :task_supervisor_reference,
+           :task_registry_reference,
            :debug_logging,
-           :cluster_task_supervisor_registry
+           :cluster_task_supervisor_registry_reference
          ])},
         strategy: :one_for_one,
         subscribe_to: [{execution_broadcaster, max_demand: 50}]
@@ -72,20 +68,23 @@ defmodule Quantum.ExecutorSupervisor do
     end
   else
     def init(
-          %{
-            execution_broadcaster: execution_broadcaster
+          %InitOpts{
+            execution_broadcaster_reference: execution_broadcaster
           } = opts
         ) do
+      executor_opts =
+        struct!(
+          ExecutorStartOpts,
+          Map.take(opts, [
+            :task_supervisor_reference,
+            :task_registry_reference,
+            :debug_logging,
+            :cluster_task_supervisor_registry_reference
+          ])
+        )
+
       ConsumerSupervisor.init(
-        [
-          {Quantum.Executor,
-           Map.take(opts, [
-             :task_supervisor,
-             :task_registry,
-             :debug_logging,
-             :cluster_task_supervisor_registry
-           ])}
-        ],
+        [{Quantum.Executor, executor_opts}],
         strategy: :one_for_one,
         subscribe_to: [{execution_broadcaster, max_demand: 50}]
       )
@@ -93,49 +92,24 @@ defmodule Quantum.ExecutorSupervisor do
   end
 
   @doc false
-  @spec child_spec(
-          {
-            GenServer.server(),
-            GenServer.server(),
-            GenServer.server(),
-            GenServer.server(),
-            boolean()
-          }
-          | {
-              GenServer.server(),
-              GenServer.server(),
-              GenServer.server(),
-              GenServer.server(),
-              GenServer.server(),
-              boolean()
-            }
-        ) :: Supervisor.child_spec()
+  @spec child_spec({
+          GenServer.server(),
+          GenServer.server(),
+          GenServer.server(),
+          GenServer.server(),
+          boolean()
+        }) :: Supervisor.child_spec()
   def child_spec({name, execution_broadcaster, task_supervisor, task_registry, debug_logging}),
     do:
-      child_spec(
-        {name, execution_broadcaster, task_supervisor, task_registry, nil, debug_logging}
-      )
+      child_spec(%StartOpts{
+        name: name,
+        execution_broadcaster_reference: execution_broadcaster,
+        task_supervisor_reference: task_supervisor,
+        task_registry_reference: task_registry,
+        cluster_task_supervisor_registry_reference: nil,
+        debug_logging: debug_logging
+      })
 
-  def child_spec(
-        {name, execution_broadcaster, task_supervisor, task_registry,
-         cluster_task_supervisor_registry, debug_logging}
-      ) do
-    %{
-      super([])
-      | start: {
-          __MODULE__,
-          :start_link,
-          [
-            %{
-              name: name,
-              execution_broadcaster: execution_broadcaster,
-              task_supervisor: task_supervisor,
-              task_registry: task_registry,
-              cluster_task_supervisor_registry: cluster_task_supervisor_registry,
-              debug_logging: debug_logging
-            }
-          ]
-        }
-    }
-  end
+  @spec child_spec(StartOpts.t()) :: Supervisor.child_spec()
+  def child_spec(%StartOpts{} = opts), do: super(opts)
 end
diff --git a/lib/quantum/executor_supervisor/init_opts.ex b/lib/quantum/executor_supervisor/init_opts.ex
new file mode 100644
index 0000000..5cc7c6a
--- /dev/null
+++ b/lib/quantum/executor_supervisor/init_opts.ex
@@ -0,0 +1,22 @@
+defmodule Quantum.ExecutorSupervisor.InitOpts do
+  @moduledoc false
+
+  # Init Options for Quantum.ExecutorSupervisor
+
+  @type t :: %__MODULE__{
+          execution_broadcaster_reference: GenServer.server(),
+          task_supervisor_reference: GenServer.server(),
+          task_registry_reference: GenServer.server(),
+          cluster_task_supervisor_registry_reference: GenServer.server() | nil,
+          debug_logging: boolean
+        }
+
+  @enforce_keys [
+    :execution_broadcaster_reference,
+    :task_supervisor_reference,
+    :task_registry_reference,
+    :cluster_task_supervisor_registry_reference,
+    :debug_logging
+  ]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/executor_supervisor/start_opts.ex b/lib/quantum/executor_supervisor/start_opts.ex
new file mode 100644
index 0000000..927246c
--- /dev/null
+++ b/lib/quantum/executor_supervisor/start_opts.ex
@@ -0,0 +1,24 @@
+defmodule Quantum.ExecutorSupervisor.StartOpts do
+  @moduledoc false
+
+  # Start Options for Quantum.ExecutorSupervisor
+
+  @type t :: %__MODULE__{
+          name: GenServer.server(),
+          execution_broadcaster_reference: GenServer.server(),
+          task_supervisor_reference: GenServer.server(),
+          task_registry_reference: GenServer.server(),
+          cluster_task_supervisor_registry_reference: GenServer.server() | nil,
+          debug_logging: boolean()
+        }
+
+  @enforce_keys [
+    :name,
+    :execution_broadcaster_reference,
+    :task_supervisor_reference,
+    :task_registry_reference,
+    :cluster_task_supervisor_registry_reference,
+    :debug_logging
+  ]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/job_broadcaster.ex b/lib/quantum/job_broadcaster.ex
index 0aef084..7a77fe5 100644
--- a/lib/quantum/job_broadcaster.ex
+++ b/lib/quantum/job_broadcaster.ex
@@ -9,39 +9,57 @@ defmodule Quantum.JobBroadcaster do
 
   alias Quantum.{Job, Scheduler}
   alias Quantum.Storage.Adapter
+  alias __MODULE__.{InitOpts, StartOpts, State}
+
+  @type event :: {:add, Job.t()} | {:remove, Job.t()}
 
   @doc """
   Start Job Broadcaster
-
-  ### Arguments
-
-   * `name` - Name of the GenStage
-   * `jobs` - Array of `Quantum.Job`
-
   """
+  @spec start_link(StartOpts.t()) :: GenServer.on_start()
+  def start_link(%StartOpts{name: name} = opts) do
+    GenStage.start_link(
+      __MODULE__,
+      struct!(InitOpts, Map.take(opts, [:jobs, :storage, :scheduler, :debug_logging])),
+      name: name
+    )
+  end
+
   @spec start_link(GenServer.server(), [Job.t()], Adapter, Scheduler, boolean()) ::
           GenServer.on_start()
   def start_link(name, jobs, storage, scheduler, debug_logging) do
-    GenStage.start_link(__MODULE__, {jobs, storage, scheduler, debug_logging}, name: name)
+    start_link(%StartOpts{
+      name: name,
+      jobs: jobs,
+      storage: storage,
+      scheduler: scheduler,
+      debug_logging: debug_logging
+    })
   end
 
   @doc false
+  @spec child_spec(StartOpts.t()) :: Supervisor.child_spec()
+  def child_spec(%StartOpts{} = opts), do: super(opts)
+
   @spec child_spec({Keyword.t() | GenServer.server(), [Job.t()], Adapter, Scheduler, boolean()}) ::
           Supervisor.child_spec()
-  def child_spec({opts, jobs, storage, scheduler, debug_logging}) when is_list(opts) do
-    %{
-      super(opts)
-      | start:
-          {__MODULE__, :start_link,
-           [Keyword.fetch!(opts, :name), jobs, storage, scheduler, debug_logging]}
-    }
-  end
-
   def child_spec({name, jobs, storage, scheduler, debug_logging}),
-    do: child_spec({[name: name], jobs, storage, scheduler, debug_logging})
+    do:
+      child_spec(%StartOpts{
+        name: name,
+        jobs: jobs,
+        storage: storage,
+        scheduler: scheduler,
+        debug_logging: debug_logging
+      })
 
   @doc false
-  def init({jobs, storage, scheduler, debug_logging}) do
+  def init(%InitOpts{
+        jobs: jobs,
+        storage: storage,
+        scheduler: scheduler,
+        debug_logging: debug_logging
+      }) do
     effective_jobs =
       scheduler
       |> storage.jobs()
@@ -63,18 +81,17 @@ defmodule Quantum.JobBroadcaster do
           storage_jobs
       end
 
-    state = %{
-      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
-    }
-
-    {:producer, state}
+    {:producer,
+     %State{
+       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
+     }}
   end
 
-  def handle_demand(demand, %{buffer: buffer} = state) do
+  def handle_demand(demand, %State{buffer: buffer} = state) do
     {to_send, remaining} = Enum.split(buffer, demand)
 
     {:noreply, to_send, %{state | buffer: remaining}}
@@ -82,7 +99,7 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:add, %Job{state: :active, name: job_name} = job},
-        %{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
+        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
           state
       ) do
     debug_logging &&
@@ -97,7 +114,7 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:add, %Job{state: :inactive, name: job_name} = job},
-        %{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
+        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
           state
       ) do
     debug_logging &&
@@ -112,7 +129,7 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:delete, name},
-        %{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
+        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
           state
       ) do
     debug_logging &&
@@ -138,7 +155,7 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:change_state, name, new_state},
-        %{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
+        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
           state
       ) do
     debug_logging &&
@@ -170,7 +187,7 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         :delete_all,
-        %{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
+        %State{jobs: jobs, storage: storage, scheduler: scheduler, debug_logging: debug_logging} =
           state
       ) do
     debug_logging &&
@@ -187,7 +204,7 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:swarm, :end_handoff, {handoff_jobs, handoff_buffer}},
-        %{
+        %State{
           jobs: jobs,
           buffer: buffer
         } = state
@@ -202,7 +219,7 @@ defmodule Quantum.JobBroadcaster do
 
   def handle_cast(
         {:swarm, :resolve_conflict, {handoff_jobs, handoff_buffer}},
-        %{
+        %State{
           jobs: jobs,
           buffer: buffer
         } = state
@@ -215,12 +232,13 @@ defmodule Quantum.JobBroadcaster do
     {:noreply, %{state | jobs: new_jobs, buffer: buffer ++ handoff_buffer}}
   end
 
-  def handle_call(:jobs, _, %{jobs: jobs} = state), do: {:reply, Map.to_list(jobs), [], state}
+  def handle_call(:jobs, _, %State{jobs: jobs} = state),
+    do: {:reply, Map.to_list(jobs), [], state}
 
-  def handle_call({:find_job, name}, _, %{jobs: jobs} = state),
+  def handle_call({:find_job, name}, _, %State{jobs: jobs} = state),
     do: {:reply, Map.get(jobs, name), [], state}
 
-  def handle_call({:swarm, :begin_handoff}, _from, %{jobs: jobs, buffer: buffer} = state) do
+  def handle_call({:swarm, :begin_handoff}, _from, %State{jobs: jobs, buffer: buffer} = state) do
     Logger.info(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Handing of state to other cluster node"
     end)
diff --git a/lib/quantum/job_broadcaster/init_opts.ex b/lib/quantum/job_broadcaster/init_opts.ex
new file mode 100644
index 0000000..d79eef8
--- /dev/null
+++ b/lib/quantum/job_broadcaster/init_opts.ex
@@ -0,0 +1,17 @@
+defmodule Quantum.JobBroadcaster.InitOpts do
+  @moduledoc false
+
+  # Init Options for Quantum.JobBroadcaster
+
+  alias Quantum.{Job, Scheduler, Storage.Adapter}
+
+  @type t :: %__MODULE__{
+          jobs: [Job.t()],
+          storage: Adapter,
+          scheduler: Scheduler,
+          debug_logging: boolean
+        }
+
+  @enforce_keys [:jobs, :storage, :scheduler, :debug_logging]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/job_broadcaster/start_opts.ex b/lib/quantum/job_broadcaster/start_opts.ex
new file mode 100644
index 0000000..187f96d
--- /dev/null
+++ b/lib/quantum/job_broadcaster/start_opts.ex
@@ -0,0 +1,18 @@
+defmodule Quantum.JobBroadcaster.StartOpts do
+  @moduledoc false
+
+  # Start Options for Quantum.JobBroadcaster
+
+  alias Quantum.{Job, Scheduler, Storage.Adapter}
+
+  @type t :: %__MODULE__{
+          name: GenServer.server(),
+          jobs: [Job.t()],
+          storage: Adapter,
+          scheduler: Scheduler,
+          debug_logging: boolean
+        }
+
+  @enforce_keys [:name, :jobs, :storage, :scheduler, :debug_logging]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/job_broadcaster/state.ex b/lib/quantum/job_broadcaster/state.ex
new file mode 100644
index 0000000..fddc54e
--- /dev/null
+++ b/lib/quantum/job_broadcaster/state.ex
@@ -0,0 +1,18 @@
+defmodule Quantum.JobBroadcaster.State do
+  @moduledoc false
+
+  # State of Quantum.JobBroadcaster
+
+  alias Quantum.{Job, JobBroadcaster, Scheduler, Storage}
+
+  @type t :: %__MODULE__{
+          jobs: %{optional(Job.name()) => Job.t()},
+          buffer: JobBroadcaster.event(),
+          storage: Storage,
+          scheduler: Scheduler,
+          debug_logging: boolean()
+        }
+
+  @enforce_keys [:jobs, :buffer, :storage, :scheduler, :debug_logging]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/scheduler.ex b/lib/quantum/scheduler.ex
index 72c9513..af948e8 100644
--- a/lib/quantum/scheduler.ex
+++ b/lib/quantum/scheduler.ex
@@ -66,10 +66,7 @@ defmodule Quantum.Scheduler do
       defp __job_broadcaster__ do
         configuration = config()
 
-        job_broadcaster =
-          configuration
-          |> Keyword.fetch!(:job_broadcaster)
-          |> Keyword.fetch!(:name)
+        job_broadcaster = Keyword.fetch!(configuration, :job_broadcaster_reference)
 
         __job_broadcaster__(job_broadcaster, configuration)
       end
diff --git a/lib/quantum/supervisor.ex b/lib/quantum/supervisor.ex
index d32c93a..c29b47a 100644
--- a/lib/quantum/supervisor.ex
+++ b/lib/quantum/supervisor.ex
@@ -3,6 +3,20 @@ defmodule Quantum.Supervisor do
 
   use Supervisor
 
+  alias Quantum.ClusterTaskSupervisorRegistry
+
+  alias Quantum.ClusterTaskSupervisorRegistry.StartOpts,
+    as: ClusterTaskSupervisorRegistryStartOpts
+
+  alias Quantum.ExecutionBroadcaster
+  alias Quantum.ExecutionBroadcaster.StartOpts, as: ExecutionBroadcasterStartOpts
+  alias Quantum.ExecutorSupervisor
+  alias Quantum.ExecutorSupervisor.StartOpts, as: ExecutorSupervisorStartOpts
+  alias Quantum.JobBroadcaster
+  alias Quantum.JobBroadcaster.StartOpts, as: JobBroadcasterStartOpts
+  alias Quantum.TaskRegistry
+  alias Quantum.TaskRegistry.StartOpts, as: TaskRegistryStartOpts
+
   @doc """
   Starts the quantum supervisor.
   """
@@ -17,58 +31,89 @@ defmodule Quantum.Supervisor do
   def init({quantum, otp_app, opts}) do
     opts = Quantum.runtime_config(quantum, otp_app, opts)
     opts = quantum_init(quantum, opts)
+    opts = Enum.into(opts, %{})
+
+    %{
+      task_registry_name: task_registry_name,
+      job_broadcaster_name: job_broadcaster_name,
+      execution_broadcaster_name: execution_broadcaster_name,
+      cluster_task_supervisor_registry_name: cluster_task_supervisor_registry_name,
+      task_supervisor_name: task_supervisor_name,
+      executor_supervisor_name: executor_supervisor_name,
+      global: global
+    } = opts
+
+    task_registry_opts = %TaskRegistryStartOpts{name: task_registry_name}
+
+    job_broadcaster_opts =
+      struct!(
+        JobBroadcasterStartOpts,
+        opts
+        |> Map.take([:jobs, :storage, :scheduler, :debug_logging])
+        |> Map.put(:name, job_broadcaster_name)
+      )
+
+    cluster_task_supervisor_registry_opts =
+      struct!(
+        ClusterTaskSupervisorRegistryStartOpts,
+        opts
+        |> Map.take([:task_supervisor_reference])
+        |> Map.put(:name, cluster_task_supervisor_registry_name)
+      )
+
+    execution_broadcaster_opts =
+      struct!(
+        ExecutionBroadcasterStartOpts,
+        opts
+        |> Map.take([:job_broadcaster_reference, :storage, :scheduler, :debug_logging])
+        |> Map.put(:name, execution_broadcaster_name)
+      )
+
+    executor_supervisor_opts =
+      struct!(
+        ExecutorSupervisorStartOpts,
+        opts
+        |> Map.take([
+          :execution_broadcaster_reference,
+          :task_supervisor_reference,
+          :task_registry_reference,
+          :cluster_task_supervisor_registry_reference,
+          :debug_logging
+        ])
+        |> Map.put(:name, executor_supervisor_name)
+      )
+
+    workers =
+      if global do
+        %{start: {TaskRegistry, f, a}} = TaskRegistry.child_spec(task_registry_opts)
+        Swarm.register_name(task_registry_name, TaskRegistry, f, a, 15_000)
+
+        %{start: {JobBroadcaster, f, a}} = JobBroadcaster.child_spec(job_broadcaster_opts)
+        Swarm.register_name(job_broadcaster_name, JobBroadcaster, f, a)
 
-    task_registry_opts = Keyword.fetch!(opts, :task_registry)
-    task_registry_name = Keyword.fetch!(task_registry_opts, :name)
+        %{start: {ExecutionBroadcaster, f, a}} =
+          ExecutionBroadcaster.child_spec(execution_broadcaster_opts)
 
-    job_broadcaster_opts = Keyword.fetch!(opts, :job_broadcaster)
-    job_broadcaster_name = Keyword.fetch!(job_broadcaster_opts, :name)
+        Swarm.register_name(execution_broadcaster_name, ExecutionBroadcaster, f, a)
 
-    execution_broadcaster_opts = Keyword.fetch!(opts, :execution_broadcaster)
-    execution_broadcaster_name = Keyword.fetch!(execution_broadcaster_opts, :name)
+        [
+          {Task.Supervisor, [name: task_supervisor_name]},
+          {ClusterTaskSupervisorRegistry, cluster_task_supervisor_registry_opts},
+          {ExecutorSupervisor, executor_supervisor_opts}
+        ]
+      else
+        [
+          {Task.Supervisor, [name: task_supervisor_name]},
+          {ClusterTaskSupervisorRegistry, cluster_task_supervisor_registry_opts},
+          {TaskRegistry, task_registry_opts},
+          {JobBroadcaster, job_broadcaster_opts},
+          {ExecutionBroadcaster, execution_broadcaster_opts},
+          {ExecutorSupervisor, executor_supervisor_opts}
+        ]
+      end
 
     Supervisor.init(
-      [
-        {Task.Supervisor, [name: Keyword.get(opts, :task_supervisor)]},
-        {Quantum.ClusterTaskSupervisorRegistry,
-         task_supervisor: Keyword.get(opts, :task_supervisor),
-         name: Keyword.get(opts, :cluster_task_supervisor_registry)},
-        {
-          Quantum.TaskRegistry,
-          task_registry_opts
-        },
-        {
-          Quantum.JobBroadcaster,
-          {
-            job_broadcaster_opts,
-            Keyword.fetch!(opts, :jobs),
-            Keyword.fetch!(opts, :storage),
-            Keyword.fetch!(opts, :quantum),
-            Keyword.fetch!(opts, :debug_logging)
-          }
-        },
-        {
-          Quantum.ExecutionBroadcaster,
-          {
-            execution_broadcaster_opts,
-            job_broadcaster_name,
-            Keyword.fetch!(opts, :storage),
-            Keyword.fetch!(opts, :quantum),
-            Keyword.fetch!(opts, :debug_logging)
-          }
-        },
-        {
-          Quantum.ExecutorSupervisor,
-          {
-            Keyword.fetch!(opts, :executor_supervisor),
-            execution_broadcaster_name,
-            Keyword.fetch!(opts, :task_supervisor),
-            task_registry_name,
-            Keyword.get(opts, :cluster_task_supervisor_registry),
-            Keyword.fetch!(opts, :debug_logging)
-          }
-        }
-      ],
+      workers,
       strategy: :rest_for_one
     )
   end
diff --git a/lib/quantum/task_registry.ex b/lib/quantum/task_registry.ex
index 0fcefdb..be5b61f 100644
--- a/lib/quantum/task_registry.ex
+++ b/lib/quantum/task_registry.ex
@@ -7,34 +7,22 @@ defmodule Quantum.TaskRegistry do
 
   require Logger
 
+  alias __MODULE__.{InitOpts, StartOpts, State}
+
   @doc """
   Start the registry
+  """
 
-  ### Options
-
-    * `name` - Name of the registry
+  @spec start_link(StartOpts.t()) :: GenServer.on_start()
+  def start_link(%StartOpts{name: name}) do
+    GenServer.start_link(__MODULE__, %InitOpts{}, name: name)
+  end
 
-  """
   @spec start_link(GenServer.server()) :: GenServer.on_start()
   def start_link(name) do
-    GenServer.start_link(__MODULE__, %{}, name: name)
-  end
-
-  @doc false
-  @spec child_spec(Keyword.t() | GenServer.server()) :: Supervisor.child_spec()
-  def child_spec(opts) when is_list(opts) do
-    %{
-      super(opts)
-      | start: {
-          __MODULE__,
-          :start_link,
-          [Keyword.fetch!(opts, :name)]
-        }
-    }
+    start_link(%StartOpts{name: name})
   end
 
-  def child_spec(name), do: child_spec(name: name)
-
   @doc """
   Mark a task as Running
 
@@ -100,31 +88,37 @@ defmodule Quantum.TaskRegistry do
   end
 
   @doc false
-  def init(args) do
-    {:ok, args}
+  def init(%InitOpts{}) do
+    {:ok, %State{running_tasks: %{}}}
   end
 
   @doc false
-  def handle_call({:running, task, node}, _caller, state) do
-    if Enum.member?(Map.get(state, task, []), node) do
+  def handle_call({:running, task, node}, _caller, %State{running_tasks: running_tasks} = state) do
+    if Enum.member?(Map.get(running_tasks, task, []), node) do
       {:reply, :already_running, state}
     else
-      {:reply, :marked_running, Map.update(state, task, [node], &[node | &1])}
+      {:reply, :marked_running,
+       %{state | running_tasks: Map.update(running_tasks, task, [node], &[node | &1])}}
     end
   end
 
   @doc false
-  def handle_call({:is_running?, task}, _caller, state) do
-    if Enum.empty?(Map.get(state, task, [])) do
-      {:reply, false, state}
-    else
-      {:reply, true, state}
+  def handle_call({:is_running?, task}, _caller, %State{running_tasks: running_tasks} = state) do
+    case running_tasks do
+      %{^task => [_ | _]} ->
+        {:reply, true, state}
+
+      %{^task => []} ->
+        {:reply, false, state}
+
+      %{} ->
+        {:reply, false, state}
     end
   end
 
   @doc false
-  def handle_call(:any_running?, _caller, state) do
-    if Enum.empty?(state) do
+  def handle_call(:any_running?, _caller, %State{running_tasks: running_tasks} = state) do
+    if Enum.empty?(running_tasks) do
       {:reply, false, state}
     else
       {:reply, true, state}
@@ -140,33 +134,41 @@ defmodule Quantum.TaskRegistry do
   end
 
   @doc false
-  def handle_cast({:finished, task, node}, state) do
-    state = Map.update(state, task, [], &(&1 -- [node]))
-
-    state =
-      if Enum.empty?(Map.fetch!(state, task)) do
-        Map.delete(state, task)
-      else
-        state
+  def handle_cast({:finished, task, node}, %State{running_tasks: running_tasks} = state) do
+    running_tasks =
+      running_tasks
+      |> Map.update(task, [], &(&1 -- [node]))
+      |> case do
+        %{^task => []} = still_running_tasks ->
+          Map.delete(still_running_tasks, task)
+
+        still_running_tasks ->
+          still_running_tasks
       end
 
-    {:noreply, state}
+    {:noreply, %{state | running_tasks: running_tasks}}
   end
 
-  def handle_cast({:swarm, :end_handoff, handoff_state}, state) do
+  def handle_cast(
+        {:swarm, :end_handoff, %State{running_tasks: handoff_tasks}},
+        %State{running_tasks: running_tasks} = state
+      ) do
     Logger.info(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Incorperating state from other cluster node"
     end)
 
-    {:noreply, merge_states(state, handoff_state)}
+    {:noreply, %{state | running_tasks: merge_states(running_tasks, handoff_tasks)}}
   end
 
-  def handle_cast({:swarm, :resolve_conflict, handoff_state}, state) do
+  def handle_cast(
+        {:swarm, :resolve_conflict, %State{running_tasks: handoff_tasks}},
+        %State{running_tasks: running_tasks} = state
+      ) do
     Logger.info(fn ->
       "[#{inspect(Node.self())}][#{__MODULE__}] Incorperating conflict state from other cluster node"
     end)
 
-    {:noreply, merge_states(state, handoff_state)}
+    {:noreply, %{state | running_tasks: merge_states(running_tasks, handoff_tasks)}}
   end
 
   defp merge_states(state1, state2) do
diff --git a/lib/quantum/task_registry/init_opts.ex b/lib/quantum/task_registry/init_opts.ex
new file mode 100644
index 0000000..9784a4c
--- /dev/null
+++ b/lib/quantum/task_registry/init_opts.ex
@@ -0,0 +1,10 @@
+defmodule Quantum.TaskRegistry.InitOpts do
+  @moduledoc false
+
+  # Init Options for Quantum.TaskRegistry
+
+  @type t :: %__MODULE__{}
+
+  @enforce_keys []
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/task_registry/start_opts.ex b/lib/quantum/task_registry/start_opts.ex
new file mode 100644
index 0000000..255dc97
--- /dev/null
+++ b/lib/quantum/task_registry/start_opts.ex
@@ -0,0 +1,12 @@
+defmodule Quantum.TaskRegistry.StartOpts do
+  @moduledoc false
+
+  # Start Options for Quantum.TaskRegistry
+
+  @type t :: %__MODULE__{
+          name: GenServer.server()
+        }
+
+  @enforce_keys [:name]
+  defstruct @enforce_keys
+end
diff --git a/lib/quantum/task_registry/state.ex b/lib/quantum/task_registry/state.ex
new file mode 100644
index 0000000..f603a4f
--- /dev/null
+++ b/lib/quantum/task_registry/state.ex
@@ -0,0 +1,14 @@
+defmodule Quantum.TaskRegistry.State do
+  @moduledoc false
+
+  # State of Quantum.TaskRegistry
+
+  alias Quantum.Job
+
+  @type t :: %__MODULE__{
+          running_tasks: %{optional(Job.name()) => [Node.t()]}
+        }
+
+  @enforce_keys [:running_tasks]
+  defstruct @enforce_keys
+end


Test patch:
diff --git a/test/quantum/cluster_task_supervisor_registry_test.exs b/test/quantum/cluster_task_supervisor_registry_test.exs
index 8f9b828..3682622 100644
--- a/test/quantum/cluster_task_supervisor_registry_test.exs
+++ b/test/quantum/cluster_task_supervisor_registry_test.exs
@@ -4,6 +4,7 @@ defmodule Quantum.ClusterTaskSupervisorRegistryTest do
   use ExUnit.Case
 
   alias Quantum.ClusterTaskSupervisorRegistry
+  alias Quantum.ClusterTaskSupervisorRegistry.StartOpts
 
   test "should register name", %{test: test} do
     {:ok, task_supervisor_pid} = start_supervised({Task.Supervisor, name: test})
@@ -11,9 +12,11 @@ defmodule Quantum.ClusterTaskSupervisorRegistryTest do
     {:ok, registry_pid} =
       start_supervised(
         {ClusterTaskSupervisorRegistry,
-         name: Module.concat([__MODULE__, test, Registry]),
-         task_supervisor: test,
-         group_name: Module.concat([__MODULE__, test, Group])}
+         %StartOpts{
+           name: Module.concat([__MODULE__, test, Registry]),
+           task_supervisor_reference: test,
+           group_name: Module.concat([__MODULE__, test, Group])
+         }}
       )
 
     Process.sleep(5_000)
@@ -35,11 +38,11 @@ defmodule Quantum.ClusterTaskSupervisorRegistryTest do
 
       send(
         test_pid,
-        ClusterTaskSupervisorRegistry.start_link(
+        ClusterTaskSupervisorRegistry.start_link(%StartOpts{
           name: Module.concat([__MODULE__, test, Registry]),
-          task_supervisor: test,
+          task_supervisor_reference: test,
           group_name: Module.concat([__MODULE__, test, Group])
-        )
+        })
       )
     end)
 
diff --git a/test/quantum/execution_broadcaster_test.exs b/test/quantum/execution_broadcaster_test.exs
index 86b5e67..e79e581 100644
--- a/test/quantum/execution_broadcaster_test.exs
+++ b/test/quantum/execution_broadcaster_test.exs
@@ -7,7 +7,7 @@ defmodule Quantum.ExecutionBroadcasterTest do
   import ExUnit.CaptureLog
   import Quantum.CaptureLogExtend
 
-  alias Quantum.{ExecutionBroadcaster, Job}
+  alias Quantum.{ExecutionBroadcaster, ExecutionBroadcaster.State, Job}
   alias Quantum.Storage.Test, as: TestStorage
   alias Quantum.{TestConsumer, TestProducer}
 
@@ -225,7 +225,14 @@ defmodule Quantum.ExecutionBroadcasterTest do
     end
 
     test "DST creates no problems", %{debug_logging: debug_logging} do
-      state = %{jobs: [], time: ~N[2018-03-25 00:59:01], timer: nil, debug_logging: debug_logging}
+      state = %State{
+        jobs: [],
+        time: ~N[2018-03-25 00:59:01],
+        timer: nil,
+        debug_logging: debug_logging,
+        storage: TestStorage,
+        scheduler: TestScheduler
+      }
 
       job =
         TestScheduler.new_job()