             n-messages (f event)
             state @pipeline-state]
         (if uuid
+          ;; UUID has been preassigned to sentinel value
           (if-not (get-in state [:learned-sentinel input])
+            ;; Don't have a cached value for the sentinel leader
             (do (vote-for-sentinel-leader log task-id input uuid)
                 (let [node (sentinel-node-name task-id input)
                       learned (:leader (extensions/read-chunk log :sentinel node))

                                             (and (= n-messages (count decompressed))
                                                  (= learned uuid))))))]
                  (if (= learned uuid)
                    {:onyx.core/tail-batch? (and (learned-all-sentinels? event successor)
                                                 (drained-all-inputs? event successor))
                     :onyx.core/requeue? true
                     :onyx.core/decompressed filtered-segments}
                    {:onyx.core/tail-batch? false
                     :onyx.core/requeue? false
                     :onyx.core/decompressed filtered-segments})))
            (let [successor (swap! pipeline-state assoc-in [:drained-inputs input]
                                   (and (= n-messages (count decompressed))
                                        (= (get-in state [:learned-sentinel input]) uuid)))]
                (if (= (get-in state [:learned-sentinel input]) uuid)
                  {:onyx.core/tail-batch? (and (learned-all-sentinels? event successor)
                                               (drained-all-inputs? event successor))
                   :onyx.core/requeue? true
                   :onyx.core/decompressed filtered-segments}
                  {:onyx.core/tail-batch? false
                   :onyx.core/requeue? false
                   :onyx.core/decompressed filtered-segments})))
          {:onyx.core/tail-batch? false
           :onyx.core/requeue? true
           :onyx.core/decompressed filtered-segments})))
    {:onyx.core/tail-batch? false
     :onyx.core/requeue? false}))


    :as event} f]
  (if (= (last decompressed) :done)
    (if (= (:onyx/type (:onyx.core/task-map event)) :input)
      ;; Input task, delegate out to its implementation for what "done" means
      (let [n-messages (f event)]
        {:onyx.core/tail-batch? (= n-messages (count decompressed))
         :onyx.core/requeue? true
