         replica-version (m/replica-version messenger)
         epoch (m/epoch messenger)]
     (let [coordinates {:tenancy-id tenancy-id :job-id job-id :replica-version replica-version :epoch epoch} 
-          _ (assert (:write-version checkpoint))
           next-write-version (write-coordinate (:write-version checkpoint) log tenancy-id job-id coordinates)]
       (-> state
           (complete-job! job-id)

                                        :remaining (m/publishers new-messenger)
                                        :opts {:recover-coordinates coordinates}})
                (update :checkpoint merge {:epoch 0})
                (assoc :curr-replica new-replica
                       :messenger new-messenger)))

          :else
          (assoc state :curr-replica new-replica))))

                       :epoch checkpointed-epoch}
          ;; get the next version of the zk node, so we can detect when there are other writers
          write-version (:write-version checkpoint)
          next-write-version (if write-coordinate?
                               (write-coordinate write-version log tenancy-id job-id coordinates)
                               write-version)
          checkpoint? (first (shuffle [true #_false]))
          messenger (m/set-epoch! messenger (inc (m/epoch messenger)))]
      (-> state
          (update :checkpoint merge {:initiated? true
                                     :epoch (if checkpoint? (m/epoch messenger) (:epoch checkpoint))
                                     :write-version next-write-version})
          (update :barrier merge {:scheduled false
                                  :offering? true
                                  :remaining (m/publishers messenger)
                                  :opts {:completed? (:sealing? (:job state))
                                         :checkpoint? checkpoint?}})
          (assoc :messenger messenger)))))

(defn shutdown [{:keys [peer-config log workflow-depth job-id messenger] :as state}]
  (assoc state :messenger (component/stop messenger)))
