   (empty? (remove #(= :done (:message %))
                   messages)))
 
-(defn completed? [batch pending-messages read-ch]
+(defn completed? [batch pending-messages read-ch retry-ch]
   (and (all-done? (vals @pending-messages))
        (all-done? batch)
        (zero? (count (.buf read-ch)))
+       (zero? (count (.buf retry-ch)))
        (or (not (empty? @pending-messages))
            (not (empty? batch)))))
 

                       top-acked-tx
                       pending-txes
                       read-ch
                       commit-ch
                       shutdown-ch)))


                          (>!! read-ch e))))]

    {:datomic/read-ch read-ch
     :datomic/retry-ch retry-ch
     :datomic/shutdown-ch shutdown-ch
     :datomic/commit-ch commit-ch
     :datomic/producer-ch producer-ch
