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

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

  (pending?

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

  (pending?
