   (retry-segment
     [_ event segment-id]
     (when-let [msg (get @pending-messages segment-id)]
-      (>!! read-ch (assoc msg :id (random-uuid))))
+      (>!!-safe read-ch (assoc msg :id (random-uuid))))
     (swap! pending-messages dissoc segment-id))
 
   (pending?

                                           (recur tx-index next-backoff)))
                                       :shutdown))]
                          (if-not (= exit :shutdown)
                            (>!! read-ch (t/input (random-uuid) :done))))
                        (catch Exception e
                          ;; feedback exception to read-batch
                          (>!! read-ch e))))]

    {:datomic/read-ch read-ch
     :datomic/shutdown-ch shutdown-ch

                                  (if success?
                                    (recur (inc chunk-index)
                                           (seq (drop datoms-per-segment datoms)))))))
                            (>!!-safe ch (t/input (random-uuid) :done))
                            (catch Exception e
                              ;; feedback exception to read-batch
                              (>!!-safe ch e))))]

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