                         (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)

  (set-replica-version [messenger replica-version]
    (-> messenger 
        (assoc-in [:replica-version peer-id] replica-version)
        (m/set-epoch 0)))

  (replica-version [messenger]

      (let [new-messenger (reduce (fn [m _]
                                    (let [subscriber (first (messenger->subscriptions m))
                                          {:keys [message ticket] :as result} (take-messages m subscriber)] 
                                      (debug (:peer-id messenger) "MSG:" 
                                             (if message message "nil") 
                                             "New sub:" (subscriber->str (:subscriber result)))
                                      (cond-> m
                                        ticket (set-ticket (:src-peer-id subscriber) (:dst-task-id subscriber) ticket)
                                        true (update-first-subscriber (constantly (:subscriber result)))
