   [{:keys [inbound-ch] :as messenger-buffer}]
   (while (first (alts!! [inbound-ch] :default false))))
 
+(defrecord TaskState [timeout-pool links])
+
 (defrecord TaskLifeCycle
     [id log messenger-buffer messenger job-id task-id replica peer-replica-view restart-ch
      kill-ch outbox-ch seal-resp-ch completion-ch opts task-kill-ch]

            input-retry-timeout (arg-or-default :onyx/input-retry-timeout catalog-entry) 
            pending-timeout (arg-or-default :onyx/pending-timeout catalog-entry) 
            r-seq (rsc/create-r-seq pending-timeout input-retry-timeout)
            state (atom {:timeout-pool r-seq})

            _ (taoensso.timbre/info (format "[%s] Warming up Task LifeCycle for job %s, task %s" id job-id (:name task)))
            _ (validate-pending-timeout pending-timeout opts)

           (map :id (:onyx.core/batch event))))
  event)

(defn process-sentinel [task-type pipeline event]
  (if (and (= task-type :input) 
           (sentinel-found? event))
    (do (if (p-ext/drained? pipeline event)
          (complete-job event)
          (p-ext/retry-message pipeline event (sentinel-id event)))
        (update-in event
                   [:onyx.core/batch]
                   (fn [batch]
                     (remove (fn [v] (= :done (:message v)))
                             batch))))
    event))

(defn collect-next-segments [f input]
