       blocks
       (fn test-block
         [block]
-        (let [id (:id block)
-              hex (multihash/hex id)]
-          (cond
-            ; Ignore any blocks which don't match the algorithm.
-            (and algorithm (not= algorithm (:algorithm id)))
-            (d/success-deferred true)
-
-            ; Drop blocks until an id later than `after`.
-            (and after (pos? (compare after hex)))
-            (d/success-deferred true)
-
-            ; Terminate the stream if block is later than `before` or `limit`
-            ; blocks have already been returned.
-            (or (and before (neg? (compare before hex)))
-                (and (pos-int? limit) (< limit (swap! counter inc))))
-            (do (s/close! out)
-                (d/success-deferred false))
-
-            ; Otherwise, pass the block along.
-            :else
-            (s/put! out block))))
+        (if (instance? Throwable block)
+          ; Propagate error on the stream.
+          (do (s/put! out block)
+              (s/close! out)
+              (d/success-deferred false))
+          ; Determine if block matches query criteria.
+          (let [id (:id block)
+                hex (multihash/hex id)]
+            (cond
+              ; Ignore any blocks which don't match the algorithm.
+              (and algorithm (not= algorithm (:algorithm id)))
+              (d/success-deferred true)
+
+              ; Drop blocks until an id later than `after`.
+              (and after (pos? (compare after hex)))
+              (d/success-deferred true)
+
+              ; Terminate the stream if block is later than `before` or `limit`
+              ; blocks have already been returned.
+              (or (and before (neg? (compare before hex)))
+                  (and (pos-int? limit) (< limit (swap! counter inc))))
+              (do (s/close! out)
+                  (d/success-deferred false))
+
+              ; Otherwise, pass the block along.
+              :else
+              (s/put! out block)))))
         out
         {:description {:op "select-blocks"}})
     (s/source-only out)))

          (fn remove-drained
            [inputs]
            (remove #(identical? ::drained (second %)) inputs))
          (fn find-next
            [inputs]
            (if (empty? inputs)
              ; Every input is drained.
              (do (s/close! out) true)
              ; Determine the next block to output.
              (let [earliest (first (sort-by :id (map second inputs)))]
                (d/chain
                  (s/put! out earliest)
                  (fn check-put
                    [result]
                    (if result
                      ; Remove any blocks matching the one emitted.
                      (d/recur (mapv (fn remove-earliest
                                       [[input head :as pair]]
                                       (if (= (:id earliest) (:id head))
                                         [input nil]
                                         pair))
                                     inputs))
                      ; Out was closed on us.
                      false))))))))
      (s/source-only out))))



              (s/drain-into source out)
              #(s/close! out))

            ; Source threw an error; propagate it.
            (instance? Throwable s)
            (d/finally
              (s/put! out s)
              #(s/close! out))

            ; Dest threw an error; propagate it.
            (instance? Throwable d)
            (d/finally
              (s/put! out d)
              #(s/close! out))

            ; Block is present in both streams; drop and continue.
            (= (:id s) (:id d))
            (d/recur nil nil)
