                                        denorm-op
                                        opts))
                              (st/buffer (or (:buffer-size opts) 25))
-                             (st/map only-error)
-                             (st/filter identity))]
+                             (st/map only-error))]
 
-           maybe-err (st/take! trerrs)]
+           ;; consumes the whole stream, returns the first error
+           ;; or nil if no errors
+           maybe-err (st/reduce (fn [v i] (or v i)) nil trerrs)]
 
       ;; if there are errors, return the first as an exemplar
       (if (nil? maybe-err)

        denorm-vals (->> (:denormalize denorm-rel)
                         (map (fn [[scol tcol]]
                                [tcol (get source-record scol)]))
                         (into {}))
        ]
    (case denorm-op

      :upsert

  "given a Deferred keep only errors, otherwise returning Deferred<nil>"
  [dv]
  (-> dv
      (d/chain (fn [v] (when (instance? Throwable v) v)))))

(s/defn denormalize-rel
  "denormalizes a single relationship"
