                   (->> (range max-segments)
                        (keep (fn [_] (first (alts!! [read-ch timeout-ch] :priority true))))))]
       (doseq [m batch]
-        (let [message (:message m)]
-          (when-not (= message :done)
-            (swap! top-tx max (:t message))
-            (swap! pending-txes conj (:t message))))
+        (feedback-producer-exception! m)
+        (update-top-txes! m top-tx pending-txes)
         (swap! pending-messages assoc (:id m) m))
-      (when (and (all-done? (vals @pending-messages))
-                 (all-done? batch)
-                 (zero? (count (.buf read-ch)))
-                 (or (not (empty? @pending-messages))
-                     (not (empty? batch))))
+      (when (completed? batch pending-messages read-ch)
         (when-not (:checkpoint/key (:onyx.core/task-map event))
           (>!! commit-ch {:status :complete}))
         (reset! drained? true))

          batch (->> (range max-segments)
                     (keep (fn [_] (first (alts!! [read-ch timeout-ch] :priority true)))))]
      (doseq [m batch]
        (when-let [chunk-index (:chunk-index m)]
          (swap! top-chunk-index max chunk-index)
          (swap! pending-chunk-indices conj chunk-index))
        (swap! pending-messages assoc (:id m) m))
    (when (and (all-done? (vals @pending-messages))
               (all-done? batch)
               (or (not (empty? @pending-messages))
                   (not (empty? batch))))
      (>!! commit-ch {:status :complete})
      (reset! drained? true))
    {:onyx.core/batch batch}))

                                           (seq (drop datoms-per-segment datoms)))))))
                            (>!! ch (t/input (random-uuid) :done))
                            (catch Exception e
                              ;; feedback exception to read-batch
                              (>!! ch e))))]

        {:datomic/read-ch ch
         :datomic/commit-ch (:commit-ch pipeline)
