                          :onyx.core/params (or (get (:fn-params opts) task) [])
                          :onyx.core/queue queue
                          :onyx.core/sync sync
-                         :onyx.core/blocked? (atom {:try-seal? false :seal? false :complete? false})}
+                         :onyx.core/pipeline-state (atom {:try-seal? false :seal? false :complete? false})}
 
           pipeline-data (assoc pipeline-data :onyx.core/queue (extensions/optimize-concurrently queue pipeline-data))
           pipeline-data (merge pipeline-data (l-ext/inject-lifecycle-resources* pipeline-data))]

        (let [path (:path response)
              seal? (extensions/read-node sync path)]
          (if seal?
            (do (swap! blocked? assoc :try-seal? true :seal? true)
                (merge event
                       (p-ext/seal-resource event)
                       {:onyx.core/sealed? true}))
            (do (swap! blocked? assoc :try-seal? true :seal? false)
                (merge event {:onyx.core/sealed? false}))))))))

(defn munge-complete-task
  [{:keys [onyx.core/sync onyx.core/completion-node
           onyx.core/cooldown-node onyx.core/blocked? onyx.core/sealed?]
    :as event}]
  (let [state @blocked?]
    (if (or (:complete? state) (not (:try-seal? state)))
      (assoc event :onyx.core/complete-success? false)
      (let [complete-response-ch (chan)]


(defn munge-seal-resource
  [{:keys [onyx.core/sync onyx.core/exhaust-node onyx.core/seal-node
           onyx.core/pipeline-state] :as event}]
  (if (:try-seal? @pipeline-state)
    (merge event {:onyx.core/sealed? false})
    (let [seal-response-ch (chan)]
      (extensions/on-change sync seal-node #(>!! seal-response-ch %))
