       (#{:input :function} task-type)         (conj {:lifecycle :offer-barriers
                                                      :fn offer-barriers
                                                      :blockable? true})
-      (#{:input} task-type)                   (conj {:lifecycle :recover-pipeline-input 
-                                                     :fn recover-pipeline-input})
-      windowed?                               (conj {:lifecycle :recover-windows-state 
-                                                     :fn recover-windows-state})
+      (#{:input} task-type)                   (conj {:lifecycle :recover-input 
+                                                     :fn recover-input})
+      windowed?                               (conj {:lifecycle :recover-state 
+                                                     :fn recover-state})
       (#{:input :function :output} task-type) (conj {:lifecycle :next-iteration
                                                      :fn next-iteration})
       (#{:input} task-type)                   (conj {:lifecycle :input-poll-barriers
                                                      :fn input-poll-barriers})
       (#{:input :function} task-type)         (conj {:lifecycle :prepare-offer-barriers
                                                      :fn prepare-offer-barriers})
-      ;; prepare-offer-barriers must come before offer-barriers
+      ;; TODO: double check that checkpoint doesn't occur immediately after recovery
+      (#{:input} task-type)                   (conj {:lifecycle :checkpoint-input
+                                                     :fn checkpoint-input
+                                                     :blockable? true})
+      windowed?                               (conj {:lifecycle :checkpoint-state
+                                                     :fn checkpoint-state
+                                                     :blockable? true})
       (#{:input :function} task-type)         (conj {:lifecycle :offer-barriers
                                                      :fn offer-barriers
                                                      :blockable? true})

        (-> state 
            (set-context! {:barrier-opts {:completed? all-barriers-completed?}
                           :publishers (m/publishers messenger)})
            (checkpoint-input!)
            (checkpoint-state!)
            (advance)))
      (goto-next-batch! state))))


        (set-pipeline! next-pipeline)
        (advance))))

(defn recover-state
  [state]
  (let [{:keys [recover] :as context} (get-context state)
        {:keys [log-prefix task-map windows triggers] :as event} (get-event state)
