                                          :checkpoint? checkpoint?}})
           (assoc :messenger messenger)))))
 
-(defn shutdown [{:keys [peer-config log workflow-depth job-id messenger] :as state}]
+(defn shutdown [{:keys [messenger] :as state}]
   (assoc state :messenger (component/stop messenger)))
 
-(defn initialise-state [{:keys [log job-id peer-config] :as state}]
-  (let [{:keys [onyx/tenancy-id]} peer-config
-        write-version (assume-checkpoint-coordinate log tenancy-id job-id)] 
-    (assert write-version)
+(defn initialise-state [{:keys [log job-id tenancy-id] :as state}]
+  (let [write-version (assume-checkpoint-coordinate log tenancy-id job-id)] 
     (-> state 
         (assoc-in [:checkpoint :write-version] write-version)
         (assoc :last-heartbeat-time (System/nanoTime)))))

       (merge-statuses)))

(defn periodic-barrier 
  [{:keys [peer-config workflow-depth log curr-replica job-id messenger barrier checkpoint] :as state}]
  (if (:offering? barrier)
    ;; No op because hasn't finished emitting last barrier, wait again
    state
    (let [{:keys [onyx/tenancy-id]} peer-config
          job-sealed? (boolean (get-in curr-replica [:completed-job-coordinates job-id]))
          checkpointed-epoch (:min-epoch (merged-statuses messenger))
          write-coordinate? (> checkpointed-epoch 0)
          coordinates {:tenancy-id tenancy-id

             :shutdown-ch shutdown-ch
             :messenger messenger
             :coordinator-thread (start-coordinator! 
                                   {:tenancy-id (:onyx/tenancy-id peer-config)
                                    :workflow-depth workflow-depth
                                    :resume-point resume-point
                                    :log log
                                    :peer-config peer-config 
