           (and sealing? (not (checkpointing? (:checkpoint state) status)))
           (complete-job state)
 
-          (or completed? sealing?
+          (or completed? 
+              sealing?
               (not= (m/replica-version messenger) (:replica-version status))
               (not= (m/epoch messenger) (:min-epoch status)))
           (do

                                         :checkpoint? checkpoint?}})
          (assoc :messenger messenger)))))

(defn shutdown [{:keys [peer-config log workflow-depth job-id 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)
    (-> state 
        (assoc-in [:checkpoint :write-version] write-version)
        (assoc :last-heartbeat-time (System/nanoTime)))))

       (merge-statuses)))

(defn periodic-barrier 
  [{:keys [tenancy-id 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 [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
