                         (is-next-barrier? messenger m))))
          first)))
 
+(defn barrier->str [barrier]
+  (str "B: " [(:replica-version barrier) (:epoch barrier)]))
+
+(defn subscriber->str [subscriber]
+  (str "S: " [(:src-peer-id subscriber) (:dst-task-id subscriber)]
+       " STATE: " (barrier->str (:barrier subscriber))))
+
 (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)) 
+        skip-to-barrier? (or (nil? (:barrier subscriber))
+                             (and (< (inc (:position subscriber)) ticket) 
                                   (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)]
+          (let [max-index (if (nil? (:barrier subscriber)) 
+                            (dec (count (get-messages messenger subscriber)))
+                            ticket)
+                [position next-barrier] (next-barrier messenger subscriber max-index)]
+            (debug (:peer-id messenger) (subscriber->str subscriber) (barrier->str next-barrier))
             {:ticket (if position 
                        (max ticket (inc position))
                        ticket)


  (all-barriers-seen? 
    [messenger]
    (info "CALL BARRIES SEEN"
          (vec (remove #(found-next-barrier? messenger %) 
                    (messenger->subscriptions messenger)))
          )
    (empty? (remove #(found-next-barrier? messenger %) 
                    (messenger->subscriptions messenger))))

  (ack-barrier
    [messenger]
    (as-> messenger mn 
      (m/next-epoch 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))
      (update-in mn
                 [:subscriptions peer-id] 
                 (fn [ss]


(defn take-ack [messenger {:keys [src-peer-id dst-task-id position] :as subscriber}]
  (let [messages (get-in messenger [:message-state src-peer-id dst-task-id])
        [idx ack] (->> messages
                       (map (fn [idx msg] [idx msg]) (range))
                       (drop (inc position))
                       (filter (fn [[idx ack]]
                                 (= (:replica-version ack) (m/replica-version messenger))))
                       (first))]
    (if ack 
      (assoc subscriber :barrier-ack ack :position idx)
      subscriber)))

(defn set-barrier-emitted [subscriber]
