Repository: clj-commons/aleph (clojure)

Issue:
Support cancellation of HTTP requests
# Problem

At the moment it is not possible to cancel HTTP requests (be it in-flight or during connection establishment). The only option users have is to just drop the response deferred on the floor and have it GC'ed. However, this doesn't free up the underlying connection until the response is complete. Combined with the fact that there is no default `request-timeout`, this can lead to resource exhaustion. 

Manifold's idiom for cancelling a deferred is to put it into an error state but that only affects *chained* deferreds downstream from the "cancelled" one (see also https://github.com/clj-commons/manifold/issues/167). Since the response deferred returned by `aleph.http/request` sits at the very end of the deferred chain, putting it into an error state has no repercussion on any of the upstream deferreds nor on the underlying connection. 

This also means that placing a timeout on a response deferred to limit the overall request duration (i.e. connection setup, request transmission and response reception) doesn't have the effect a user might expect:

```clojure
@(-> (http/get "...") (d/timeout! 1000))
```

As explained above, this will simply put the response deferred into an error state when the timeout expires but the underlying request operation will still proceed, tying up any resources it has acquired until done.

# Solutions

The most elegant solution would be to change Manifold to also cancel any upstream deferreds which feed into a "cancelled" deferred (if there are no others left). However, [according to Zach](https://github.com/clj-commons/manifold/issues/166#issuecomment-459902958), this cannot be implemented in the current architecture of Manifold and would require a "major reengineering".

A more tractable solution would be to explicitly cancel any upstream operations when the response deferred is set into an error state.

Pull request:
This patch changes `aleph.http/request` so that setting the response deferred to an error status will terminate an in-flight request. This allows e.g. for `d/timeout!` to be used without potentially leaking connections.

For convenient explicit cancellation, we provide `aleph.http/cancel-request!`. It puts the given response deferred into error state with an instance of the new `aleph.utils.RequestCancellationException`.

Closes #712.

Review notes:
* ~The patch uses the approach outlined in https://github.com/clj-commons/manifold/issues/167#issuecomment-457996116 for propagating cancellation.~  (not quite, it was merely an inspiration)
* Diff best viewed [without whitespace changes](https://github.com/clj-commons/aleph/pull/714/files?w=1).
* ~Still marked as draft because the patch only pertains to in-flight requests. Connection establishment is not yet cancellable. Combined with the default `connection-timeout` of 60s, this can make for an unpleasant surprise :grimacing:~ -- deferred to a follow-up patch once Netty has responded.

Patch:
diff --git a/src-java/aleph/utils/RequestCancellationException.java b/src-java/aleph/utils/RequestCancellationException.java
new file mode 100644
index 00000000..948c8788
--- /dev/null
+++ b/src-java/aleph/utils/RequestCancellationException.java
@@ -0,0 +1,23 @@
+package aleph.utils;
+
+import java.util.concurrent.CancellationException;
+
+public class RequestCancellationException extends CancellationException {
+
+    public RequestCancellationException() { }
+
+    public RequestCancellationException(String message) {
+        super(message);
+    }
+
+    public RequestCancellationException(Throwable cause) {
+        super(cause.getMessage());
+        initCause(cause);
+    }
+
+    public RequestCancellationException(String message, Throwable cause) {
+        super(message);
+        initCause(cause);
+    }
+
+}
diff --git a/src/aleph/http.clj b/src/aleph/http.clj
index 630647cb..f0a847e4 100644
--- a/src/aleph/http.clj
+++ b/src/aleph/http.clj
@@ -20,6 +20,7 @@
       ConnectionTimeoutException
       PoolTimeoutException
       ReadTimeoutException
+      RequestCancellationException
       RequestTimeoutException)
     (io.aleph.dirigiste Pools)
     (io.netty.handler.codec Headers)
@@ -336,6 +337,9 @@
      by [clj-http](https://github.com/dakrone/clj-http), and returns a deferred representing
      the HTTP response.  Also allows for a custom `pool` or `middleware` to be defined.
 
+     Putting the returned deferred into an error state will cancel the underlying request if it is
+     still in flight.
+
      Param key            | Description
      -------------------- | -----------------------------------------------------------------------------------------------------------------------------------------------------------------
      `connection-timeout` | timeout in milliseconds for the connection to become established
@@ -358,96 +362,106 @@
              middleware         identity
              connection-timeout 6e4}                        ;; 60 seconds
       :as   req}]
-
-    (executor/with-executor response-executor
-      ((middleware
-         (fn [req]
-           (let [k (client/req->domain req)
-                 start (System/currentTimeMillis)]
-
-             ;; acquire a connection
-             (-> (flow/acquire pool k)
-                 (maybe-timeout! pool-timeout)
-
-                 ;; pool timeout triggered
-                 (d/catch' TimeoutException
-                   (fn [^Throwable e]
-                     (d/error-deferred (PoolTimeoutException. e))))
-
-                 (d/chain'
-                   (fn [conn]
-
-                     ;; get the wrapper for the connection, which may or may not be realized yet
-                     (-> (first conn)
-                         (maybe-timeout! connection-timeout)
-
-                         ;; connection timeout triggered, dispose of the connetion
-                         (d/catch' TimeoutException
-                           (fn [^Throwable e]
-                             (log/error e "Timed out waiting for connection to be established")
-                             (flow/dispose pool k conn)
-                             (d/error-deferred (ConnectionTimeoutException. e))))
-
-                         ;; connection failed, bail out
-                         (d/catch'
-                           (fn [e]
-                             (log/error e "Connection failure")
-                             (flow/dispose pool k conn)
-                             (d/error-deferred e)))
-
-                         ;; actually make the request now
-                         (d/chain'
-                           (fn [conn']
-                             (when-not (nil? conn')
-                               (let [end (System/currentTimeMillis)]
-                                 (-> (conn' req)
-                                     (maybe-timeout! request-timeout)
-
-                                     ;; request timeout triggered, dispose of the connection
-                                     (d/catch' TimeoutException
-                                       (fn [^Throwable e]
-                                         (flow/dispose pool k conn)
-                                         (d/error-deferred (RequestTimeoutException. e))))
-
-                                     ;; request failed, dispose of the connection
-                                     (d/catch'
-                                       (fn [e]
-                                         (log/trace "Request failed. Disposing of connection...")
-                                         (flow/dispose pool k conn)
-                                         (d/error-deferred e)))
-
-                                     ;; clean up the connection
-                                     (d/chain'
-                                       (fn cleanup-conn [rsp]
-
-                                         ;; either destroy/dispose of the conn, or release it back for reuse
-                                         (-> (:aleph/destroy-conn? rsp)
-                                             (maybe-timeout! read-timeout)
-
-                                             (d/catch' TimeoutException
-                                               (fn [^Throwable e]
-                                                 (log/trace "Request timed out. Disposing of connection...")
-                                                 (flow/dispose pool k conn)
-                                                 (d/error-deferred (ReadTimeoutException. e))))
-
-                                             (d/chain'
-                                               (fn [early?]
-                                                 (if (or early?
-                                                         (not (:aleph/keep-alive? rsp))
-                                                         (<= 400 (:status rsp)))
-                                                   (do
-                                                     (log/trace "Connection finished. Disposing...")
-                                                     (flow/dispose pool k conn))
-                                                   (flow/release pool k conn)))))
-                                         (-> rsp
-                                             (dissoc :aleph/destroy-conn?)
-                                             (assoc :connection-time (- end start)))))))))
-
-                           (fn handle-response [rsp]
-                             (->> rsp
-                                  (middleware/handle-cookies req)
-                                  (middleware/handle-redirects request req)))))))))))
-       req))))
+    (let [dispose-conn! (atom (fn []))
+          result (d/deferred response-executor)
+          response (executor/with-executor response-executor
+                     ((middleware
+                       (fn [req]
+                         (let [k (client/req->domain req)
+                               start (System/currentTimeMillis)]
+
+                           ;; acquire a connection
+                           (-> (flow/acquire pool k)
+                               (maybe-timeout! pool-timeout)
+
+                               ;; pool timeout triggered
+                               (d/catch' TimeoutException
+                                         (fn [^Throwable e]
+                                           (d/error-deferred (PoolTimeoutException. e))))
+
+                               (d/chain'
+                                (fn [conn]
+                                  ;; NOTE: All error handlers below delegate disposal of the
+                                  ;; connection to the error handler on `result` which uses this
+                                  ;; function.
+                                  (reset! dispose-conn! (fn [] (flow/dispose pool k conn)))
+
+                                  (if (realized? result)
+                                    ;; to account for race condition between setting `dispose-conn!`
+                                    ;; and putting `result` into error state for cancellation
+                                    (@dispose-conn!)
+                                    ;; get the wrapper for the connection, which may or may not be realized yet
+                                    (-> (first conn)
+                                        (maybe-timeout! connection-timeout)
+
+                                        ;; connection timeout triggered
+                                        (d/catch' TimeoutException
+                                                  (fn [^Throwable e]
+                                                    (d/error-deferred (ConnectionTimeoutException. e))))
+
+                                        ;; actually make the request now
+                                        (d/chain'
+                                         (fn [conn']
+                                           (when-not (nil? conn')
+                                             (let [end (System/currentTimeMillis)]
+                                               (-> (conn' req)
+                                                   (maybe-timeout! request-timeout)
+
+                                                   ;; request timeout triggered
+                                                   (d/catch' TimeoutException
+                                                             (fn [^Throwable e]
+                                                               (d/error-deferred (RequestTimeoutException. e))))
+
+                                                   ;; clean up the connection
+                                                   (d/chain'
+                                                    (fn cleanup-conn [rsp]
+
+                                                      ;; either destroy/dispose of the conn, or release it back for reuse
+                                                      (-> (:aleph/destroy-conn? rsp)
+                                                          (maybe-timeout! read-timeout)
+
+                                                          ;; read timeout triggered
+                                                          (d/catch' TimeoutException
+                                                                    (fn [^Throwable e]
+                                                                      (log/trace "Request timed out.")
+                                                                      (d/error-deferred (ReadTimeoutException. e))))
+
+                                                          (d/chain'
+                                                           (fn [early?]
+                                                             (if (or early?
+                                                                     (not (:aleph/keep-alive? rsp))
+                                                                     (<= 400 (:status rsp)))
+                                                               (do
+                                                                 (log/trace "Connection finished. Disposing...")
+                                                                 (flow/dispose pool k conn))
+                                                               (flow/release pool k conn)))))
+                                                      (-> rsp
+                                                          (dissoc :aleph/destroy-conn?)
+                                                          (assoc :connection-time (- end start)))))))))
+
+                                         (fn handle-response [rsp]
+                                           (->> rsp
+                                                (middleware/handle-cookies req)
+                                                (middleware/handle-redirects request 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)))
+
+(defn cancel-request!
+  "Accepts a response deferred as returned by `request` and closes the underlying TCP connection. If
+  the request had already completed by the time this function is invoked, it has no effect (as per
+  Manifold deferred semantics). If cancellation succeeded, the deferred will be put into error state
+  with an `aleph.utils.RequestCancellationException` instance.
+
+  Note that the request may already have been (partially) processed by the server at the point of
+  cancellation."
+  [r]
+  (d/error! r (RequestCancellationException. "Request cancelled")))
 
 (defn- req
   ([method url]


Test patch:
diff --git a/test/aleph/http_test.clj b/test/aleph/http_test.clj
index 0e251112..23a14202 100644
--- a/test/aleph/http_test.clj
+++ b/test/aleph/http_test.clj
@@ -18,6 +18,7 @@
   (:import
     (aleph.utils
       ConnectionTimeoutException
+      RequestCancellationException
       RequestTimeoutException)
     (clojure.lang
       ExceptionInfo)
@@ -1439,6 +1440,20 @@
       (is (instance? IllegalArgumentException result))
       (is (= "use-h2c? may only be true when HTTP/2 is enabled." (ex-message result))))))
 
+(deftest test-in-flight-request-cancellation
+  (let [conn-established (promise)
+        conn-closed (promise)]
+    (with-raw-handler (fn [req]
+                        (deliver conn-established true)
+                        (s/on-closed (:body req)
+                                     (fn []
+                                       (deliver conn-closed true))))
+      (let [rsp (http-get "/")]
+        (is (= true (deref conn-established 1000 :timeout)))
+        (http/cancel-request! rsp)
+        (is (= true (deref conn-closed 1000 :timeout)))
+        (is (thrown? RequestCancellationException (deref rsp 1000 :timeout)))))))
+
 (deftest ^:leak test-leak-in-raw-stream-handler
   ;; NOTE: Expecting 2 leaks because `with-raw-handler` will run its body for both http1 and
   ;; http2. It would be nicer to put this assertion into the body but the http1 server seems to