           (if (pos? ret)
             (recur (rest pubs))
             (set-context! state (assoc context :publishers pubs))))
-        (do
-         ;; Move this into separate lifecycle?
-         (println "Unblocking for, windowed?" (:windowed-task? event) "rve" (m/replica-version messenger) (m/epoch messenger))
-         (m/unblock-subscribers! messenger)
-         (advance state))))))
+        (advance state)))))
+
+(defn unblock-subscribers [state]
+  (m/unblock-subscribers! (get-messenger state))
+  (advance state))
 
 ;; After offer barriers, have notify publishers?
 ;; Then we can unblock subscribers

        (m/unblock-subscribers! messenger)
        (when (and (m/all-barriers-completed? messenger) 
                   (not (sealed? state)))
          (-> state 
              (set-sealed! true)
              (complete-job!)))
        (m/next-epoch! messenger)
        (let [{:keys [job-id task-id slot-id log]} (get-event state)
              replica-version (m/replica-version messenger)

      (#{:input :function} task-type)         (conj {:lifecycle :offer-barriers
                                                     :fn offer-barriers
                                                     :blockable? true})
      (#{:input :function} task-type)         (conj {:lifecycle :unblock-subscribers
                                                     :fn unblock-subscribers})
      (#{:input} task-type)                   (conj {:lifecycle :recover-input 
                                                     :fn recover-input})
      windowed?                               (conj {:lifecycle :recover-state 
