               ; because non-immediate reactions are buffered til join
               (let [replica (:replica state)
                     peerless-queues (->> (:entries state)
-                                         (filter (fn [[queue-id queue]] 
+                                         (filter (fn [[queue-id {:keys [predicate queue]}]] 
                                                    (or (peerless-entry? (first queue))
-                                                       (allow-queue? replica queue))))
+                                                       ((or predicate (constantly true)) 
+                                                        replica 
+                                                        (first queue)))))
                                          (map key))
                     joined-peers (set (:peers replica))
                     selectable-peers (->> (:entries state)
-                                          (filter (fn [[peer [entry]]]
-                                                    (or (:immediate? entry)
-                                                        (contains? joined-peers peer))))
+                                          (filter (fn [[peer {:keys [queue]}]]
+                                                    (let [entry (first queue)] 
+                                                      (or (:immediate? entry)
+                                                          (contains? joined-peers peer)))))
                                           (map key)
                                           set)
                     selectable-queues (into selectable-peers peerless-queues)]

                                 (when (seq reactions)
                                   [peer-id reactions])))
                             peers)
        side-effects (keep (partial generate-side-effects entry replica new-replica diff) peers)
        new-entries (concat peer-reactions side-effects)
        ; it does not matter that multiple reactions are processed
        ; together because they may be processed interleaved depending on 
        ; the choice of peer queue being popped
        unapplied (reduce (fn [new-entries [peer-id reactions]]
                            (update-in new-entries 
                                       [peer-id] 
                                       (fn [entries] (-> entries
                                                         vec
                                                         (into reactions)
                                                         (bump-forward-immediates (:peers replica) peer-id)))))
                          entries
                          new-entries)]
    (vector new-replica diff unapplied)))

(defn apply-peer-queue-entry 
  "Applies the next log message in the selected peer's queue.
  Effectively, the next peer that wrote its message to ZK"
  [{:keys [replica message-id entries peer-choices log]} next-peer] 
  (let [peer-queue (entries next-peer)
        next-entry (first peer-queue)
        new-peer-queue (vec (rest peer-queue))
        new-entries (if (empty? new-peer-queue)
                      (dissoc entries next-peer)
                      (assoc entries next-peer new-peer-queue))
        message (assoc next-entry :message-id message-id)
        [new-replica diff updated-entries] (apply-entry replica new-entries message)] 
    {:replica new-replica


(defn generate-join-queues [peer-ids]
  (zipmap peer-ids
          (map (fn [peer] 
                 {:queue [(build-join-entry peer)]}) 
               peer-ids))) 

(defn generate-peer-ids [n]
  (map #(keyword (str "p" %))
