 (defn complete-task [sync task-node]
   (extensions/create-node sync (str task-node complete-marker)))
 
-(defn n-peers [sync peers task-node states]
-  (count
-   (filter
-    (fn [peer]
-      (let [state (:content (extensions/dereference sync peer))]
-        (and (= (:task-node state) task-node)
-             (some #{(:state state)} states))))
-    peers)))
+(defn n-peers [sync task-node]
+  (let [peers (extensions/bucket sync :peer-state)
+        counts
+        (reduce
+         (fn [all peer]
+           (let [content (:content (extensions/dereference sync peer))]
+             (if (= task-node (:task-node content))
+               (assoc all (:state content) (inc (get all (:state content) 0)))
+               all)))
+         {}
+         peers)]
+    ;; Avoid NPE on addition
+    (-> counts
+        (assoc :acking (get counts :acking 0))
+        (assoc :active (get counts :active 0))
+        (assoc :waiting (get counts :waiting 0))
+        (assoc :sealing (get counts :sealing 0)))))
 
 (defmethod extensions/mark-peer-born ZooKeeper
   [sync peer-node]

            incomplete-tasks)))

(defn sort-tasks-by-peer-count [sync tasks]
  (sort-by #(n-peers sync (extensions/bucket sync :peer-state)
                     % #{:active :acking :sealing}) (sort tasks)))

(defn next-active-task [sync job-node]
  (let [incomplete-tasks (find-incomplete-concurrent-tasks sync job-node)]

  (let [states (extensions/bucket sync :peer-state)]
    (filter
     (fn [task]
       (let [peers (n-peers sync task)]
         (>= (+ (:acking peers) (:active peers) (:sealing peers)) 1)))
     tasks)))

(defn next-inactive-task [sync job-node]
