     (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)

  (m/poll (get-messenger state))
  (advance state))

(defn checkpoint-input [state]
  (let [messenger (get-messenger state)
        pipeline (get-pipeline state)
        {:keys [job-id task-id slot-id log]} (get-event state)
        replica-version (m/replica-version messenger)
        epoch (m/epoch messenger)
        checkpoint (oi/checkpoint pipeline)] 
    (extensions/write-checkpoint log job-id replica-version epoch task-id slot-id :input checkpoint)
    (advance state)))

(defn checkpoint-state [state]
  (let [messenger (get-messenger state)
        {:keys [job-id task-id slot-id log]} (get-event state)
        replica-version (m/replica-version messenger)
        epoch (m/epoch messenger)] 
    (extensions/write-checkpoint log job-id replica-version epoch task-id slot-id 
                                 :state 
                                 (mapv ws/export-state (get-windows-state state)))
    (advance state)))

(defn prepare-offer-barriers [state]
  (let [messenger (get-messenger state)] 
