   (set-replica-version [messenger replica-version]
     (-> messenger 
         (assoc-in [:replica-version peer-id] replica-version)
+        (update-in [:subscriptions peer-id] 
+                   (fn [ss] (mapv #(assoc % :barrier-ack nil :barrier nil) ss)))
         (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)] 
                                      (cond-> m
                                        ticket (set-ticket (:src-peer-id subscriber) (:dst-task-id subscriber) ticket)
                                        true (update-first-subscriber (constantly (:subscriber result)))

    (update-in messenger [:epoch peer-id] inc))

  (receive-acks [messenger]
    (update-in messenger 
               [:subscriptions peer-id]
               (fn [ss]
                 (mapv (fn [s] (take-ack messenger s)) ss))))

  (flush-acks [messenger]
    (update-in messenger 
               [:subscriptions peer-id]
               (fn [ss]
                 (mapv #(assoc % :barrier-ack nil) ss))))

  (all-acks-seen? 
    [messenger]
    (if (empty? (remove :barrier-ack (messenger->subscriptions messenger)))
      (:barrier-ack (first (messenger->subscriptions messenger)))
      false))
