 (defn take-final-state!! [component]
   (<!! (:task-lifecycle-ch component)))
 
-(defn compile-task [{:keys [task-information job-id task-id id monitoring log
-                            replica-origin replica opts outbox-ch group-ch task-kill-flag kill-flag]}]
+(defn compile-task 
+  [{:keys [task-information job-id task-id id monitoring log replica-origin
+           replica opts outbox-ch group-ch task-kill-flag kill-flag]}]
   (let [{:keys [workflow catalog task flow-conditions resume-point 
                 windows triggers lifecycles metadata]} task-information
         log-prefix (logger/log-prefix task-information)
         task-map (find-task catalog (:name task))
         filtered-windows (vec (wc/filter-windows windows (:name task)))
         window-ids (set (map :window/id filtered-windows))
-        filtered-triggers (filterv #(window-ids (:trigger/window-id %)) triggers)
+        filtered-triggers (filterv (comp window-ids :trigger/window-id) triggers)
         _ (info log-prefix "Compiling lifecycle")]
     (assert (:onyx/tenancy-id opts) (:onyx/tenancy-id opts))
     (->> {:onyx.core/id id

            [onyx.log.entry :as entry]
            [onyx.log.replica]
            [onyx.messaging.common :as mc]
            [onyx.messaging.protocols.messenger :as m]
            [onyx.messaging.protocols.publisher :as pub]
            [onyx.messaging.protocols.subscriber :as sub]
            [onyx.messaging.protocols.status-publisher :as status-pub]
            [onyx.messaging.messenger-state :as ms]
            [onyx.monitoring.measurements :refer [emit-latency emit-latency-value]]
            [onyx.peer.constants :refer [initialize-epoch]]
            [onyx.peer.task-compile :as c]

       (remove nil?)
       (first)))

(defn wrap-lifecycle-metrics [monitoring lifecycle]
  (let [lfn (:fn lifecycle)]
    (if-let [mon-fn (get monitoring (:lifecycle lifecycle))]
      (fn [state]
        (let [start (System/nanoTime)
              next-state (lfn state)
              end (System/nanoTime)
              elapsed (unchecked-subtract end start)]
          (mon-fn next-state elapsed)
          next-state))
      lfn)))

(defn lookup-batch-start-index [lifecycles]
  ;; before-batch may be stripped, thus before or read may be first batch fn
  (int (or (lookup-lifecycle-idx lifecycles :lifecycle/before-batch)
           (lookup-lifecycle-idx lifecycles :lifecycle/read-batch))))

(defn new-state-machine [event opts messenger messenger-group coordinator] 
  (let [{:keys [onyx.core/input-plugin onyx.core/output-plugin onyx.core/monitoring]} event
        {:keys [replica-version] :as base-replica} (onyx.log.replica/starting-replica opts)
        lifecycles (filter :fn (filter-task-lifecycles event))
        names (into-array clojure.lang.Keyword (mapv :lifecycle lifecycles))
        state-fns (->> lifecycles
                       (mapv #(wrap-lifecycle-metrics monitoring %))
                       (into-array clojure.lang.IFn))
        recover-idx (int 0)
        iteration-idx (int (lookup-lifecycle-idx lifecycles :lifecycle/next-iteration))
        batch-idx (lookup-batch-start-index lifecycles)
        start-idx recover-idx
        heartbeat-ns (ms->ns (arg-or-default :onyx.peer/heartbeat-ms opts))
        idle-strategy (SleepingIdleStrategy. (arg-or-default :onyx.peer/idle-sleep-ns opts))
        window-states (c/event->windows-states event)]
    (->TaskStateMachine idle-strategy recover-idx iteration-idx batch-idx
                        (count state-fns) names state-fns start-idx false
                        false base-replica monitoring messenger messenger-group
                        coordinator input-plugin output-plugin event event
                        window-states nil replica-version initialize-epoch
                        heartbeat-ns (System/nanoTime))))

;; NOTE: currently, if task doesn't start before the liveness timeout, the peer will be killed
;; peer should probably be heartbeating here
(defn backoff-until-task-start! 
  [{:keys [onyx.core/kill-flag onyx.core/task-kill-flag onyx.core/opts] :as event} start-fn]
  (while (and (not (or @kill-flag @task-kill-flag))
