             [onyx.log.commands.common :as common]
             [onyx.extensions :as extensions]
             [onyx.peer.operation :as operation]
-            [onyx.scheduling.common-job-scheduler :as cjs]
+            [onyx.scheduling.common-job-scheduler :refer [reconfigure-cluster-workload]]
             [taoensso.timbre :refer [info] :as timbre]))
-
-(defn anticipating-coverage? [old new job-id]
-  (let [n-tasks (count (get-in new [:tasks job-id]))
-        n-tasks-covered (count (get-in new [:allocations job-id]))
-        n-volunteering (->> (:peers new)
-                            (filter #(cjs/reallocate-from-job? (:job-scheduler old) old new {:id %}))
-                            (count))]
-    (>= n-volunteering (- n-tasks n-tasks-covered))))
-
-(defn volunteer? [old new state job-id]
-  (and (cjs/reallocate-from-job? (:job-scheduler old) old new state)
-       (anticipating-coverage? old new job-id)))
 
 (defn add-site [replica {:keys [joiner peer-site]}]
   (-> replica 

                watcher (nth sorted-candidates index)]
            (-> replica
                (update-in [:prepared] merge {watcher joining-peer})
                (add-site args)))
          replica))
      (-> replica
          (update-in [:peers] conj (:joiner args))
          (update-in [:peers] vec)
          (assoc-in [:peer-state (:joiner args)] :idle)
          (add-site args)))))

(defmethod extensions/replica-diff :prepare-join-cluster
  [entry old new]

          :args {:observer (:subject diff)
                 :subject (or (get (:pairs new) (:observer diff))
                              (:observer diff))}
          :immediate? true}]))

(defmethod extensions/fire-side-effects! :prepare-join-cluster
  [{:keys [args]} old new diff state]
  (common/start-new-lifecycle
   (cond (= (:id state) (:observer diff))
         (let [ch (chan 1)]
           (extensions/on-delete (:log state) (:subject diff) ch)
           (go (when (<! ch)
                 (extensions/write-log-entry
                  (:log state)
                  {:fn :leave-cluster :args {:id (:subject diff)}}))
               (close! ch))
           (assoc state :watch-ch ch))
         (= (:id state) (:subject diff))
         (let [ch (chan 1)]
           (extensions/on-delete (:log state) (:observer diff) ch)
           (go (when (<! ch)
                 (extensions/write-log-entry
                  (:log state)
                  {:fn :leave-cluster :args {:id (:observer diff)}}))
               (close! ch))
           (assoc state :watch-ch ch))
         (= (:id state) (:instant-join diff))
         (do (extensions/open-peer-site (:messenger state)
                                        (get-in new [:peer-sites (:id state)]))
             (doseq [entry (:buffered-outbox state)]
               (>!! (:outbox-ch state) entry))
             (assoc (dissoc state :buffered-outbox) :stall-output? false))
         :else state)
   diff))
