     (update-in messenger [:epoch peer-id] inc))
 
   (receive-acks [messenger]
-    (reduce (fn [messenger _]
-              (let [subscriber (first (messenger->subscriptions messenger))]
-                (if (:barrier-ack subscriber) 
-                  messenger
-                  (update-first-subscriber messenger (constantly (take-ack messenger subscriber))))))
-            messenger
-            (messenger->subscriptions messenger)))
+    (update-in messenger 
+               [:subscriptions peer-id]
+               (fn [ss]
+                 (mapv (fn [s] (take-ack messenger s)) ss))))
 
   (flush-acks [messenger]
-    (reduce (fn [messenger _]
-              (let [subscriber (first (messenger->subscriptions messenger))]
-                (update-first-subscriber messenger (constantly (assoc subscriber :barrier-ack nil)))))
-            messenger
-            (messenger->subscriptions messenger)))
+    (update-in messenger 
+               [:subscriptions peer-id]
+               (fn [ss]
+                 (mapv #(assoc % :barrier-ack nil) ss))))
 
   (all-acks-seen? 
     [messenger]
-    (info "All acks seen? " (vec (messenger->subscriptions messenger)))
     (if (empty? (remove :barrier-ack (messenger->subscriptions messenger)))
       (:barrier-ack (first (messenger->subscriptions messenger)))
       false))

                        (is-next-barrier? messenger m))))
         first)))

(defn take-messages [messenger subscriber]
  (let [ticket (curr-ticket messenger subscriber) 
        next-ticket (inc ticket)
        message (get-message messenger subscriber ticket)
        skip-to-barrier? (and (< (inc (:position subscriber)) ticket) 
                              ;; Only skip ahead when unblocked. Otherwise we may skip over next barriers
                              (or (nil? (:barrier subscriber)) 
                                  (unblocked? messenger subscriber)))] 
    (cond skip-to-barrier?
          ;; Skip up to next barrier so we're aligned again, 
          ;; but don't move past actual messages that haven't been read
          (let [[position next-barrier] (next-barrier messenger subscriber ticket)]
            {:ticket (if position 
                       (max ticket (inc position))
                       ticket)


  (all-barriers-seen? 
    [messenger]
    (empty? (remove #(found-next-barrier? messenger %) 
                    (messenger->subscriptions messenger))))

  (ack-barrier
    [messenger]
    (as-> messenger mn 
      (reduce (fn [m p] 
                (write m p (->BarrierAck peer-id (:dst-task-id p) (m/replica-version mn) (m/epoch mn)))) 
              mn 
              (get publications peer-id))
      (m/next-epoch mn)
      (update-in mn
                 [:subscriptions peer-id] 
                 (fn [ss]
