         max-pending (or (:onyx/max-pending catalog-entry) (:onyx/max-pending defaults))
         batch-size (:onyx/batch-size catalog-entry)
         batch-timeout (or (:onyx/batch-timeout catalog-entry) (:onyx/batch-timeout defaults))
-        read-ch (chan (or (:datomic/read-buffer catalog-entry) 1000))] 
+        read-ch (chan (or (:datomic/read-buffer catalog-entry) 1000))
+        commit-ch (chan (sliding-buffer 1))] 
     (->DatomicInput (:onyx.core/log pipeline-data)
                     (:onyx.core/task-id pipeline-data)
                     max-pending batch-size batch-timeout 

            db (safe-as-of task-map conn)
            datoms-per-segment (safe-datoms-per-segment task-map)
            unroll (partial unroll-datom db)
            num-ignored (* start-index datoms-per-segment)]
        (go
          (try
            (loop [chunk-index (inc start-index)
                   datoms (seq (drop num-ignored
                                     (datoms-sequence db task-map)))]
              (when datoms 
                (>!! ch (assoc (t/input (java.util.UUID/randomUUID)
                                        {:datoms (map unroll (take datoms-per-segment datoms))})
                               :chunk-index chunk-index))
                (recur (inc chunk-index) 
                       (seq (drop datoms-per-segment datoms)))))
            (>!! ch (t/input (java.util.UUID/randomUUID) :done))
            (catch Exception e
              (fatal e))))
        {:datomic/read-ch ch
         :datomic/drained? (:drained pipeline)
         :datomic/top-chunk-index (:top-chunk-index pipeline)
         :datomic/top-acked-chunk-index (:top-acked-chunk-index pipeline)

  (ack-segment [_ _ segment-id]
    (let [chunk-index (:chunk-index (@pending-messages segment-id))]
      (swap! pending-chunk-indices disj chunk-index)
      (let [new-top-acked (highest-acked-chunk @top-acked-chunk-index @top-chunk-index @pending-chunk-indices)]
        (>!! commit-ch {:chunk-index new-top-acked :status :incomplete})
        (reset! top-acked-chunk-index new-top-acked))
      (swap! pending-messages dissoc segment-id)))

