Don't leak unconsumed deferreds in error state in HTTP client

src/aleph/http.clj

@@ -487,14 +487,17 @@
                                          (fn handle-response [rsp]
                                            (->> rsp
                                                 (middleware/handle-cookies req)
-                                                (middleware/handle-redirects request req))))))))))))
+                                                (middleware/handle-redirects request req)
+                                                (middleware/handle-error-status req))))))))))))
                       req))]
       (d/connect response result)
-      (d/catch' result
-                (fn [e]
-                  (log/trace e "Request failed. Disposing of connection...")
-                  (@dispose-conn!)
-                  (d/error-deferred e)))
+      (-> result
+          (d/catch'
+           (fn [e]
+             (when-not (:aleph/error-status? (ex-data e))
+               (log/trace e "Request failed. Disposing of connection...")
+               (@dispose-conn!)
+               true))))
       result)))
 
 (defn cancel-request!

src/aleph/http/client_middleware.clj

@@ -254,36 +254,39 @@
     (str "application/" (name type))
     type))
 
-(defn wrap-exceptions
-  "Middleware that throws response as an ExceptionInfo if the response has
-  unsuccessful status code. :throw-exceptions set to false in the request
-  disables this middleware."
-  [client]
-  (fn [req]
-    (d/let-flow' [{:keys [status body] :as rsp} (client req)]
-      (if (unexceptional-status? status)
-        rsp
-        (cond
-
-          (false? (opt req :throw-exceptions))
-          rsp
-
-          (instance? InputStream body)
-          (d/chain' (d/future (bs/to-byte-array body))
-                    (fn [body]
-                      (d/error-deferred
-                        (ex-info
-                          (str "status: " status)
-                          (assoc rsp :body (ByteArrayInputStream. body))))))
-
-          :else
-          (d/chain'
-            (s/reduce conj [] body)
-            (fn [body]
-              (d/error-deferred
-                (ex-info
-                  (str "status: " status)
-                  (assoc rsp :body (s/->source body)))))))))))
+(defn error-status-deferred [rsp]
+  (d/error-deferred
+   (ex-info
+    (str "status: " (:status rsp))
+    (assoc rsp :aleph/error-status? true))))
+
+(defn handle-error-status
+  "Wraps the response in an error deferred if it has unsuccessful status code.
+  :throw-exceptions set to false in the request disables this middleware."
+  [req rsp]
+  (let [{:keys [status body]} rsp]
+    (cond
+      (unexceptional-status? status)
+      rsp
+
+      (false? (opt req :throw-exceptions))
+      rsp
+
+      (instance? InputStream body)
+      (d/chain' (d/future (bs/to-byte-array body))
+                (fn [body]
+                  (error-status-deferred
+                   (assoc rsp :body (ByteArrayInputStream. body)))))
+
+      (nil? body)
+      (error-status-deferred rsp)
+
+      :else
+      (d/chain'
+       (s/reduce conj [] body)
+       (fn [body]
+         (error-status-deferred
+          (assoc rsp :body (s/->source body))))))))
 
 (defn wrap-method
   "Middleware converting the :method option into the :request-method option"
@@ -965,7 +968,6 @@
   by default"
   [client]
   (let [client' (-> client
-                    wrap-exceptions
                     wrap-request-timing)]
     (fn [req]
       (let [executor (ex/executor)]

test/aleph/http/client_middleware_test.clj

@@ -2,10 +2,13 @@
   (:require
    [aleph.http.client-middleware :as middleware]
    [aleph.http.schema :as schema]
+   [clojure.java.io :as io]
    [clojure.test :as t :refer [deftest is testing]]
-   [malli.core :as m]
-   [malli.generator :as mg])
+   [malli.generator :as mg]
+   [manifold.deferred :as d]
+   [manifold.stream :as s])
   (:import
+   (clojure.lang ExceptionInfo)
    (java.net URLDecoder)))
 
 (deftest test-empty-query-string
@@ -203,3 +206,45 @@
         #"Invalid spec.*:in \[:content-type\].*:value 10"
         (middleware/wrap-validation {:request-method :post
                                      :content-type 10}))))
+
+(deftest test-handle-error-status
+  (let [handle-error-status (fn [req rsp]
+                              (deref (d/chain (middleware/handle-error-status req rsp))
+                                     1000
+                                     ::timeout))]
+    (testing "response body is input stream"
+      (try
+        (let [rsp (handle-error-status {:throw-exceptions? true}
+                                       {:status 400
+                                        :body (io/input-stream (.getBytes "hello"))})]
+          (is (= ::thrown? rsp) "should have thrown"))
+        (catch Exception e
+          (is (instance? ExceptionInfo e))
+          (is (= 400 (:status (ex-data e))))
+          (is (= "hello" (some-> e ex-data :body slurp))))))
+    (testing "response body is manifold stream"
+      (try
+        (let [rsp (handle-error-status {:throw-exceptions? true}
+                                       {:status 400
+                                        :body (doto (s/stream 1)
+                                                (s/put! "hello")
+                                                (s/close!))})]
+          (is (= ::thrown? rsp) "should have thrown"))
+        (catch Exception e
+          (is (instance? ExceptionInfo e))
+          (is (= 400 (:status (ex-data e))))
+          (is (= "hello" (some-> e ex-data :body s/take! (deref 100 :timeout)))))))
+    (testing "response body is nil"
+      (try
+        (let [rsp (handle-error-status {:throw-exceptions? true}
+                                       {:status 400
+                                        :body nil})]
+          (is (= ::thrown? rsp) "should have thrown"))
+        (catch Exception e
+          (is (instance? ExceptionInfo e))
+          (is (= 400 (:status (ex-data e))))
+          (is (nil? (-> e ex-data :body))))))
+    (testing "error status but :throw-exceptions is false"
+      (let [rsp (handle-error-status {:throw-exceptions? false}
+                                     {:status 400})]
+        (is (= 400 (:status rsp)))))))

test/aleph/http_test.clj

@@ -1194,6 +1194,18 @@
                                       (d/timeout! 5e3)
                                       deref)))))
 
+(deftest test-client-throw-error-responses-as-exceptions
+  (with-http-servers (fn [_]
+                       {:status 429
+                        :body "nope"}) {}
+    (try
+      (-> (http-get "/" {:throw-exceptions? true})
+          (d/timeout! 1e3)
+          deref)
+      (is (= false true) "should have thrown an exception")
+      (catch Exception e
+        (is (= 429 (:status (ex-data e))))))))
+
 (deftest test-client-errors-handling
   (testing "writing invalid request message"
     (with-handler echo-string-handler