     (fn [state]
       (advance (set-event! state (f (get-event state)))))))
 
-(defn recover-stored-checkpoint
+(defn read-checkpoint
   [{:keys [log job-id task-id slot-id] :as event} checkpoint-type recover]
   (let [checkpointed (extensions/read-checkpoint log job-id recover task-id slot-id checkpoint-type)]
     (if-not (= :beginning checkpointed)

  (let [{:keys [recover] :as context} (get-context state)
        pipeline (get-pipeline state)
        event (get-event state)
        stored (recover-stored-checkpoint event :input recover)
        _ (assert recover)
        _ (println "RECOVER pipeline checkpoint" (:job-id event) (:task-id event) stored)
        next-pipeline (oi/recover pipeline stored)]

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