       (#{: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 

                                       :slot-id slot-id})]
    (println "Job completed" job-id task-id (:args entry))
    (info "job completed:" job-id task-id (:args entry))
    (>!! outbox-ch entry)))

(defn seal-barriers [state]
  (advance 

          (if (pos? ret)
            (recur (rest pubs))
            (set-context! state (assoc context :publishers pubs))))
        (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
