        (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}

  [{:keys [allocation-ch shutdown-ch peer-config] :as state}]
  (thread
    (try
      (let [;; FIXME: allow in job data
          ;snapshot-every-n (arg-or-default :onyx.peer/coordinator-snapshot-every-n-barriers peer-config)
            coordinator-max-sleep-ns (ms->ns (arg-or-default :onyx.peer/coordinator-max-sleep-ms peer-config))
            barrier-period-ns (ms->ns (arg-or-default :onyx.peer/coordinator-barrier-period-ms peer-config))
            heartbeat-ns (ms->ns (arg-or-default :onyx.peer/heartbeat-ms peer-config))
            result (loop [state (initialise-state state)]
                     (if-let [scheduler-event (poll! shutdown-ch)]
                       (do (shutdown (assoc state :scheduler-event scheduler-event))
                           :shutdown)
                       (if-let [new-replica (poll! allocation-ch)]
                       ;; Set up reallocation barriers. Will be sent on next recur through :offer-barriers
                         (recur (next-replica state barrier-period-ns new-replica))
                         (recur (coordinator-iteration state
                                                       coordinator-max-sleep-ns
                                                       barrier-period-ns
                                                       heartbeat-ns)))))]
        (when-not (= result :shutdown)
          (throw (Exception. "Coordinator not properly shutdown"))))
      (catch Throwable e
        (>!! (:group-ch state) [:restart-vpeer (:peer-id state)])
        (fatal e "Error in coordinator")))))

  (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
