        (catch KeeperException$BadVersionException bve
          (throw (Exception. "Coordinator failed to write coordinates.
                              This is likely due to the coordinator being quarantined, and another coordinator taking over.")))))
-
-(defn complete-job! [state job-id]
-  (>!! (:group-ch state) [:send-to-outbox {:fn :complete-job :args {:job-id job-id}}])
-  state)
 
 (defn next-replica
   [{:keys [log job-id peer-id messenger curr-replica tenancy-id] :as state}


(defn checkpointing? [{:keys [initiated? epoch] :as checkpoint} status]
  (and ;; we've initiated a snapshot
   initiated?
       ;; none of the statuses are saying they're checkpointing
   (or (:checkpointing? status)
           ;; and they are all at least up to the checkpoint barrier
           ;; which means that we are not getting a stale status
       (< (:min-epoch status) epoch))))

(defn completed-checkpoint [{:keys [checkpoint messenger job-id tenancy-id log] :as state}]
  (let [{:keys [epoch write-version]} checkpoint

  (if (:offering? barrier)
    ;; No op because hasn't finished emitting last barrier, wait again
    state
    (let [;; always checkpoints at the moment
          checkpoint? (first (shuffle [true #_false]))
          messenger (m/set-epoch! messenger (inc (m/epoch messenger)))]
      (-> state
          (update :checkpoint merge {:initiated? true
