 (defn complete-job
   [{:keys [tenancy-id log job-id messenger checkpoint group-ch] :as state}]
   (info (format "Job %s completed, and final checkpoint has finished. Writing checkpoint coordinates." job-id))
+  (swap! coordinator-log conj :complete-job)
   (let [replica-version (m/replica-version messenger)
         epoch (m/epoch messenger)]
     (let [coordinates {:tenancy-id tenancy-id 


(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


(defn offer-heartbeats
  [{:keys [messenger] :as state}]
  (swap! coordinator-log conj :offer-heartbeats)
  (run! pub/offer-heartbeat! (m/publishers messenger))
  (assoc state :last-heartbeat-time (System/nanoTime)))

(def coordinator-backoff-ms 500)

(defn offer-barriers [{:keys [messenger barrier] :as state}]
  (if (:offering? barrier)
