 
 (defn recover-state
   [state]
-  (let [{:keys [recover] :as context} (get-context state)
-        {:keys [log-prefix task-map windows triggers] :as event} (get-event state)
-        stored (recover-stored-checkpoint event :state recover)
+  (let [{:keys [log-prefix task-map windows triggers] :as event} (get-event state)
+        {:keys [recover] :as context} (get-context state)
+        stored (read-checkpoint event :state recover)
         _ (assert recover)
         recovered-windows (->> windows
                                (mapv (fn [window] (wc/resolve-window-state window triggers task-map)))

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

        (m/unblock-subscribers! messenger)
        (when (and (m/all-barriers-completed? messenger) 
                   (not (sealed? state)))
          (complete-job! state))
        (m/next-epoch! messenger)
        (let [{:keys [job-id task-id slot-id log]} (get-event state)
              replica-version (m/replica-version messenger)
