   (>!! (:group-ch state) [:send-to-outbox {:fn :complete-job :args {:job-id job-id}}])
   state)
 
-
-; WHEN all the inputs are completed? send one final "flush barrier". This will do a checkpoint, when everyone is on the min-epoch
-;; and everyone is completed, then we can complete
 (defn next-replica 
   [{:keys [peer-config log job-id peer-id messenger curr-replica] :as state} 
    barrier-period-ns

                       :epoch checkpointed-epoch}
          ;; get the next version of the zk node, so we can detect when there are other writers
          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)))]
      ;(println "SETUP BARRIER OPTS WITL SEALING" (:sealing? state))
      ;; all the inputs sealing
      (assoc state 
             :checkpointing? checkpoint?
             :checkpoint-epoch (if checkpoint? (m/epoch messenger) (:checkpoint-epoch state))

          ;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 [{:keys [messenger] :as 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")))))

(defprotocol Coordinator
  (start [this])
