session+response: work harder to ensure errors during drain are handled correctly

Author: Bogdanp

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

@@ -40,6 +40,7 @@
   [pool? (-> any/c boolean?)]
   [pool-lease (->* [pool?] [(or/c #f timeout-config?)] http-conn?)]
   [pool-release (-> pool? http-conn? void?)]
+  [pool-abandon (-> pool? http-conn? void?)]
   [pool-close! (-> pool? void?)]))
 
 (define connector/c
@@ -97,6 +98,9 @@
 (define (pool-release p c)
   (d:pool-release! (pool-impl p) c))
 
+(define (pool-abandon p c)
+  (d:pool-abandon! (pool-impl p) c))
+
 (define (pool-close! p)
   (d:pool-close! (pool-impl p))
   (log-http-easy-debug "connection pool closed"))

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

@@ -34,7 +34,7 @@
  response-output
  response-history
  (contract-out
-  [make-response (-> bytes? (listof bytes?) input-port? (listof response?) response-closer/c response-destroyer/c response?)]
+  [make-response (-> bytes? (listof bytes?) input-port? (listof response?) response-closer/c response-abandoner/c response?)]
   [response-body (-> response? bytes?)]
   [response-json (-> response? (or/c eof-object? jsexpr?))]
   [response-xexpr (-> response? xexpr?)]
@@ -44,7 +44,8 @@
   [read-response-xexpr (-> response? xexpr?)]
   [read-response-xml (-> response? document?)]
   [response-drain! (->* [response?] [(or/c #f (and/c real? (not/c negative?)))] void?)]
-  [response-close! (-> response? void?)]))
+  [response-close! (-> response? void?)]
+  [response-closed? (-> response? boolean?)]))
 
 (struct response
   (status-line
@@ -57,18 +58,18 @@
    history
    closer
    [closed? #:mutable]
-   destroyer))
+   abandoner))
 
 (define response-closer/c
   (-> response? void?))
 
-(define response-destroyer/c
+(define response-abandoner/c
   (-> response? void?))
 
 (define status-code/c
   (integer-in 100 999))
 
-(define (make-response status headers output history closer destroyer)
+(define (make-response status headers output history closer abandoner)
   (match status
     [(regexp #rx#"^HTTP/(...) ([1-9][0-9][0-9])(?: (.*))?$"
              (list status-line
@@ -89,7 +90,7 @@
         #;history history
         #;close closer
         #;closed? #f
-        #;destroyer destroyer))
+        #;abandoner abandoner))
      (retain resp)
      resp]
     [_
@@ -142,17 +143,17 @@
 (define (response-drain! r [t #f])
   (unless (response-data r)
     (parameterize-break #f
+      (log-http-easy-debug "draining response ~.s" r)
       (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))))))
+         (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)))
+        (log-http-easy-debug "timed out while draining response")
+        (raise (make-timeout-error 'request)))
       (force drain-promise))))
 
 (define (response-close! r)
@@ -162,6 +163,8 @@
   ;; time.
   (unless (response-closed? r)
     (parameterize-break #f
+      (log-http-easy-debug "closing response ~.s" r)
+      (set-response-closed?! r #t)
       (define drain-promise
         (delay/thread
          (define in (response-output r))
@@ -171,13 +174,14 @@
       (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))
+           (cond
+             [(sync/timeout 1 drain-promise)
+              (log-http-easy-debug "response closed")
+              ((response-closer r) r)]
+             [else
+              (log-http-easy-debug "timed out while closing response; abandoning connection")
+              ((response-abandoner r) r)]))))
+      (sync/enable-break close-thd)
       (force drain-promise))))
 
 

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

@@ -120,15 +120,21 @@
              (make-pool (session-conf s) connector))))))
     (pool-lease p timeouts)))
 
-(define (session-release s url c)
+(define (get-session-pool s url)
   (define k (pool-key url))
-  (define p
-    (call-with-semaphore (session-sema s)
-      (lambda ()
-        (hash-ref (session-pools s) k #f))))
-  (when p
-    (log-http-easy-debug "releasing connection to pool ~a" k)
-    (pool-release p c)))
+  (call-with-semaphore (session-sema s)
+    (lambda ()
+      (hash-ref (session-pools s) k #f))))
+
+(define (session-release s url c)
+  (define p (get-session-pool s url))
+  (log-http-easy-debug "releasing connection to pool ~s" (eq-hash-code p))
+  (pool-release p c))
+
+(define (session-abandon s url c)
+  (define p (get-session-pool s url))
+  (log-http-easy-debug "abandoning connection of pool ~s" (eq-hash-code p))
+  (pool-abandon p c))
 
 (define supplied?
   (compose1 not unsupplied-arg?))
@@ -176,19 +182,18 @@
                        (data headers)
                        (values headers data))])
       (parameterize-break #f
-        (define conn (session-lease sess u timeouts))
+        (define conn
+          (session-lease sess u timeouts))
         (define resp
           (with-handlers ([exn:break?
                            (lambda (e)
-                             (log-http-easy-warning "received break")
-                             (http-conn-close! conn)
-                             (session-release sess u conn)
+                             (log-http-easy-warning "received break during request processing; abandoning connection")
+                             (session-abandon sess u conn)
                              (raise e))]
                           [exn:fail?
                            (lambda (e)
                              (log-http-easy-warning "request failed: ~a" (exn-message e))
-                             (http-conn-close! conn)
-                             (session-release sess u conn)
+                             (session-abandon sess u conn)
                              (cond
                                [(exn:fail:http-easy? e)
                                 (log-http-easy-warning "error cannot be retried; bubbling up exception")
@@ -202,7 +207,7 @@
                                 (raise e)]))])
             (define resp-ch
               (make-channel))
-            (define thd
+            (define resp-thd
               (thread
                (lambda ()
                  (with-handlers ([exn:break? void]
@@ -223,57 +228,67 @@
                      resp-headers
                      resp-output
                      history
-                     (lambda (_)
-                       (session-release sess u conn))
-                     (lambda (_)
-                       (http-conn-close! conn))))))))
+                     (λ (_) (session-release sess u conn))
+                     (λ (_) (session-abandon sess u conn))))))))
             (with-handlers ([exn:break?
                              (lambda (e)
-                               (break-thread thd)
+                               (break-thread resp-thd)
                                (raise e))])
               (sync/enable-break
                (handle-evt
                 resp-ch
-                (lambda (r)
-                  (when (exn:fail? r)
-                    (raise r))
-                  (begin0 r
-                    (log-http-easy-debug "response: ~.s" (response-status-line r))
-                    (maybe-save-cookies! sess u (response-headers r)))))
+                (lambda (resp) ;; noqa
+                  (when (exn:fail? resp)
+                    (raise resp))
+                  (log-http-easy-debug "response: ~.s" (response-status-line resp))
+                  (maybe-save-cookies! sess u (response-headers resp))
+                  resp))
                (handle-evt
                 (make-request-timeout-evt timeouts)
                 (lambda (_)
-                  (break-thread thd)
+                  (break-thread resp-thd)
                   (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 (timeout-config-request timeouts))
-           (response-close! resp)
-           (parameterize-break enable-breaks?
-             (go dest-url
-                 #:method (case (response-status-code resp)
-                            [(301 302 303) 'get]
-                            [(307 308)     method])
-                 #:headers (hash-remove headers 'authorization)
-                 #:auth (and (same-origin? dest-url u) auth)
-                 #:history (cons resp history)
-                 #:redirects (sub1 redirects-remaining)))]
-
-          [(or close? (not stream?))
-           (response-drain! resp (timeout-config-request timeouts))
-           (response-close! resp)
-           resp]
-
-          [else
-           resp]))))
+        ;; When an error occurs at this point, a response-close! may already be in progress, so
+        ;; check that the response hasn't been closed already before abandoning the connection.
+        (with-handlers ([exn:break?
+                         (lambda (e)
+                           (log-http-easy-warning "received break during response processing")
+                           (unless (response-closed? resp)
+                             (log-http-easy-warning "abandoning connection")
+                             (session-abandon sess u conn))
+                           (raise e))]
+                        [exn:fail?
+                         (lambda (e)
+                           (log-http-easy-warning "response processing failed")
+                           (unless (response-closed? resp)
+                             (session-abandon sess u conn))
+                           (raise e))])
+          (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 (timeout-config-request timeouts))
+             (response-close! resp)
+             (parameterize-break enable-breaks?
+               (go dest-url
+                   #:method (case (response-status-code resp)
+                              [(301 302 303) 'get]
+                              [(307 308)     method])
+                   #:headers (hash-remove headers 'authorization)
+                   #:auth (and (same-origin? dest-url u) auth)
+                   #:history (cons resp history)
+                   #:redirects (sub1 redirects-remaining)))]
+            [(or close? (not stream?))
+             (response-drain! resp (timeout-config-request timeouts))
+             (response-close! resp)
+             resp]
+            [else
+             ;; No more exceptions to handle at this point, so it is safe to register a will
+             ;; executor for the response and return it.
+             (will-register executor resp response-close!)
+             resp])))))
 
   (go (->url urlish)))
 

http-easy-lib/info.rkt

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

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

@@ -419,7 +419,27 @@
           (parameterize ([current-session (make-session)])
             (check-equal? (response-body (get (format "http://127.0.0.1:~a" port))) #"ok")
             ;; https://github.com/Bogdanp/racket-http-easy/issues/25
-            (check-equal? (response-body (get (string->url/literal (format "http://127.0.0.1:~a/a%2Bb.mp3" port)))) #"ok"))))))
+            (check-equal? (response-body (get (string->url/literal (format "http://127.0.0.1:~a/a%2Bb.mp3" port)))) #"ok")))))
+
+     (test-case "response-close! on a server that won't respond in time"
+       (define sema (make-semaphore))
+       (call-with-tcp-server
+        (lambda (_lines out)
+          (fprintf out "HTTP/1.1 200 OK\r\n")
+          (fprintf out "Content-Length: 1000\r\n")
+          (fprintf out "\r\n")
+          (flush-output out)
+          (semaphore-wait sema))
+        (lambda (port)
+          (parameterize ([current-session (make-session)])
+            (define resp (get #:stream? #t (format "http://127.0.0.1:~a" port)))
+            (check-exn
+             #rx"input port is closed"
+             (lambda ()
+               (response-close! resp)))
+            (semaphore-post sema)
+            (sync (system-idle-evt))
+            (session-close! (current-session)))))))
 
     (test-suite
      "custom port"
@@ -510,28 +530,33 @@
 
      (test-case "breaking is safe"
        (define sema (make-semaphore))
-       (call-with-web-server
-        (lambda (_req)
-          (response/output
-           (lambda (out)
-             (semaphore-wait sema)
-             (displayln "hello, world!" out))))
-        (lambda (addr)
-          (parameterize ([current-session
-                          (make-session
-                           #:pool-config
-                           (make-pool-config
-                            #:max-size 1))])
-            (define thd
-              (thread
-               (lambda ()
-                 (with-handlers ([exn:break? void])
-                   (get addr)))))
-            (sync (system-idle-evt))
-            (break-thread thd)
-            (semaphore-post sema)
-            (semaphore-post sema)
-            (check-not-false (get addr))))))))))
+       (for ([_ (in-range 10)])
+         (call-with-web-server
+          (lambda (_req)
+            (response/output
+             (lambda (out)
+               (semaphore-wait sema)
+               (displayln "hello, world!" out))))
+          (lambda (addr)
+            (parameterize ([current-session
+                            (make-session
+                             #:pool-config
+                             (make-pool-config
+                              #:max-size 1))])
+              (define broken?-sema
+                (make-semaphore))
+              (define thd
+                (thread
+                 (lambda ()
+                   (with-handlers ([exn:break? void])
+                     (get addr))
+                   (semaphore-post broken?-sema))))
+              (sync (system-idle-evt))
+              (break-thread thd)
+              (semaphore-post sema)
+              (semaphore-post sema)
+              (semaphore-wait broken?-sema)
+              (check-not-false (get addr)))))))))))
 
 (module+ test
   (require rackunit/text-ui)