response: limit how long a response can take to close

Previously, closing a response would wait on the response data port to
get drained, but that meant a buggy (or adversarial) server could
block the close perpetually. Likewise, draining a connection during
redirect or when #:close? was #t had the same issue.

This change reworks closing and draining to have timeouts and to
behave as reasonably as they can in the face of such responses, by
still draining the data for compliant servers and timing out after one
second on close and after the response timeout config on drain.

Author: Bogdanp

http-easy-lib/http-easy/private/response.rkt

@@ -7,7 +7,9 @@
          racket/lazy-require
          racket/match
          racket/port
+         racket/promise
          "common.rkt"
+         "error.rkt"
          "logger.rkt"
          "port.rkt")
 
@@ -32,7 +34,7 @@
  response-output
  response-history
  (contract-out
-  [make-response (-> bytes? (listof bytes?) input-port? (listof response?) response-closer/c response?)]
+  [make-response (-> bytes? (listof bytes?) input-port? (listof response?) response-closer/c response-destroyer/c response?)]
   [response-body (-> response? bytes?)]
   [response-json (-> response? (or/c eof-object? jsexpr?))]
   [response-xexpr (-> response? xexpr?)]
@@ -41,12 +43,11 @@
   [read-response-json (-> response? (or/c eof-object? jsexpr?))]
   [read-response-xexpr (-> response? xexpr?)]
   [read-response-xml (-> response? document?)]
-  [response-drain! (-> response? void?)]
+  [response-drain! (->* [response?] [(or/c #f (and/c real? (not/c negative?)))] void?)]
   [response-close! (-> response? void?)]))
 
 (struct response
-  (sema
-   status-line
+  (status-line
    http-version
    status-code
    status-message
@@ -55,15 +56,19 @@
    [data #:mutable]
    history
    closer
-   [closed? #:mutable]))
+   [closed? #:mutable]
+   destroyer))
 
 (define response-closer/c
   (-> response? void?))
 
+(define response-destroyer/c
+  (-> response? void?))
+
 (define status-code/c
   (integer-in 100 999))
 
-(define (make-response status headers output history closer)
+(define (make-response status headers output history closer destroyer)
   (match status
     [(regexp #rx#"^HTTP/(...) ([1-9][0-9][0-9])(?: (.*))?$"
              (list status-line
@@ -72,21 +77,21 @@
                    status-message))
      (define-values (retaining-output retain)
        (make-retaining-input-port output))
-     (define the-resp
-       (response (make-semaphore 1)
-                 status-line
-                 http-version
-                 status-code
-                 (or status-message #"")
-                 headers
-                 retaining-output
-                 #f
-                 history
-                 closer
-                 #f))
-     (begin0 the-resp
-       (retain the-resp))]
-
+     (define resp
+       (response
+        #;status-linse status-line
+        #;http-version http-version
+        #;status-code status-code
+        #;status-message (or status-message #"")
+        #;headers headers
+        #;output retaining-output
+        #;data #f
+        #;history history
+        #;close closer
+        #;closed? #f
+        #;destroyer destroyer))
+     (retain resp)
+     resp]
     [_
      (raise-argument-error 'status "a valid status line" status)]))
 
@@ -134,27 +139,46 @@
 (define (read-response-xml r)
   (read-xml/document (response-output r)))
 
-(define (response-drain! r)
-  (call-with-semaphore (response-sema r)
-    (lambda ()
-      (unless (response-data r)
-        (define inp (response-output r))
-        (unless (port-closed? inp)
-          (define data (port->bytes inp))
-          (set-response-data! r data)
-          (close-input-port inp))))))
+(define (response-drain! r [t #f])
+  (unless (response-data r)
+    (parameterize-break #f
+      (define drain-promise
+        (delay/thread
+         (with-handlers ([exn:break? void])
+           (parameterize-break #t
+             (define in (response-output r))
+             (unless (port-closed? in)
+               (define data (port->bytes in))
+               (set-response-data! r data)
+               (close-input-port in))))))
+      (unless (sync/timeout/enable-break t drain-promise)
+        (raise (make-timeout-error 'drain)))
+      (force drain-promise))))
 
 (define (response-close! r)
-  (call-with-semaphore (response-sema r)
-    (lambda ()
-      (unless (response-closed? r)
-        (define inp (response-output r))
-        (unless (port-closed? inp)
-          (copy-port inp (open-output-nowhere))
-          (close-input-port inp))
-        ((response-closer r) r)
-        (set-response-closed?! r #t)
-        (log-http-easy-debug "response closed")))))
+  ;; In order to reuse the connection, we need to drain the data port,
+  ;; but draining the data port might block indefinitely, so drain with
+  ;; a timeout and close the connection if the data cannot be drained in
+  ;; time.
+  (unless (response-closed? r)
+    (parameterize-break #f
+      (define drain-promise
+        (delay/thread
+         (define in (response-output r))
+         (unless (port-closed? in)
+           (copy-port in (open-output-nowhere))
+           (close-input-port in))))
+      (define close-thd
+        (thread
+         (lambda ()
+           (unless (sync/timeout 1 drain-promise)
+             (log-http-easy-warning "timed out while closing response")
+             ((response-destroyer r) r))
+           ((response-closer r) r)
+           (log-http-easy-debug "response closed"))))
+      (set-response-closed?! r #t)
+      (sync/enable-break (thread-dead-evt close-thd))
+      (force drain-promise))))
 
 
 ;; match expanders ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

http-easy-lib/http-easy/private/session.rkt

@@ -224,7 +224,9 @@
                      resp-output
                      history
                      (lambda (_)
-                       (session-release sess u conn))))))))
+                       (session-release sess u conn))
+                     (lambda (_)
+                       (http-conn-close! conn))))))))
             (with-handlers ([exn:break?
                              (lambda (e)
                                (break-thread thd)
@@ -245,12 +247,15 @@
                   (log-http-easy-warning "request timed out~n  method: ~s~n  url: ~.s" method urlish)
                   (raise (make-timeout-error 'request))))))))
 
+        ;; Register executor early in case the calls to response-drain!
+        ;; raise an exception. Closing the response twice is a no-op.
+        (will-register executor resp response-close!)
         (cond
           [(and (positive? redirects-remaining) (redirect? resp))
            (define location (bytes->string/utf-8 (response-headers-ref resp 'location)))
            (define dest-url (ensure-absolute-url u location))
            (log-http-easy-debug "following ~s redirect to ~s" (response-status-code resp) location)
-           (response-drain! resp)
+           (response-drain! resp (timeout-config-request timeouts))
            (response-close! resp)
            (parameterize-break enable-breaks?
              (go dest-url
@@ -263,13 +268,12 @@
                  #:redirects (sub1 redirects-remaining)))]
 
           [(or close? (not stream?))
-           (begin0 resp
-             (response-drain! resp)
-             (response-close! resp))]
+           (response-drain! resp (timeout-config-request timeouts))
+           (response-close! resp)
+           resp]
 
           [else
-           (begin0 resp
-             (will-register executor resp response-close!))]))))
+           resp]))))
 
   (go (->url urlish)))
 

http-easy-lib/http-easy/private/timeout.rkt

@@ -9,11 +9,12 @@
  timeout-config-connect
  timeout-config-request
  (contract-out
-  [make-timeout-config (->* ()
-                            (#:lease timeout/c
-                             #:connect timeout/c
-                             #:request timeout/c)
-                            timeout-config?)]
+  [make-timeout-config
+   (->* []
+        [#:lease timeout/c
+         #:connect timeout/c
+         #:request timeout/c]
+        timeout-config?)]
   [make-request-timeout-evt (-> timeout-config? evt?)]))
 
 (define timeout/c
@@ -22,9 +23,10 @@
 (struct timeout-config (lease connect request)
   #:transparent)
 
-(define (make-timeout-config #:lease [lease 5]
-                             #:connect [connect 5]
-                             #:request [request 30])
+(define (make-timeout-config
+         #:lease [lease 5]
+         #:connect [connect 5]
+         #:request [request 30])
   (timeout-config lease connect request))
 
 (define (make-request-timeout-evt t)

http-easy-lib/info.rkt

@@ -1,7 +1,7 @@
 #lang info
 
 (define license 'BSD-3-Clause)
-(define version "0.8.6")
+(define version "0.9")
 (define collection "net")
 (define deps
   '(["base" #:version "8.1.0.4"]

http-easy-test/net/http-easy/private/response.rkt

@@ -12,8 +12,9 @@
          #:headers [headers null]
          #:body [body (open-input-bytes #"")]
          #:history [history null]
-         #:closer [closer void])
-  (make-response status headers body history closer))
+         #:closer [closer void]
+         #:destroyer [destroyer void])
+  (make-response status headers body history closer destroyer))
 
 (define response-tests
   (test-suite

http-easy/http-easy.scrbl

@@ -568,8 +568,15 @@ scheme and url-encode the path to the socket as the host.
   Equivalent to @racket[(read-xml/document (response-output r))].
 }
 
-@defproc[(response-drain! [r response?]) void?]{
-  Drains @racket[r]'s output port.
+@defproc[(response-drain! [r response?]
+                          [t (or/c #f (and/c real? (not/c negative?))) #f]) void?]{
+  Drains @racket[r]'s output port. When @racket[t] is provided, a
+  timeout error is raised if draining the response takes more than
+  @racket[t] seconds.
+
+  @history[
+   #:changed "0.9" @elem{Added the @racket[#t] argument.}
+  ]
 }
 
 @defproc[(response-close! [r response?]) void?]{