     (if-not (= :beginning checkpointed)
       checkpointed)))
 
-(defn recover-pipeline-input [state]
+(defn recover-input [state]
   (let [{:keys [recover] :as context} (get-context state)
         pipeline (get-pipeline state)
         event (get-event state)

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

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

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

