Try to better handle/avoid http related failures

I'm seeing "Resource temporarily unavailable, try again" errors from GnuTLS,
mostly around the file uploads I think.

I'm not sure what's going on here, but it seems to happen when using multiple
threads in parallel. Anyway, this commit uses some mutexes to avoid uploading
files in parallel, and also improves error handling generally. I'm pretty sure
this isn't sufficient to fix the issue, but I could be looking in completely
the wrong place for the problem.

1 files changed, 102 insertions, 67 deletions

diff --git a/guix-build-coordinator/agent-messaging/http.scm b/guix-build-coordinator/agent-messaging/http.scm
index 117745a..5ad0a4b 100644
--- a/guix-build-coordinator/agent-messaging/http.scm
+++ b/guix-build-coordinator/agent-messaging/http.scm
@@ -23,6 +23,7 @@
   #:use-module (srfi srfi-19)
   #:use-module (ice-9 match)
   #:use-module (ice-9 format)
+  #:use-module (ice-9 threads)
   #:use-module (ice-9 exceptions)
   #:use-module (ice-9 textual-ports)
   #:use-module (ice-9 binary-ports)
@@ -445,6 +446,29 @@ port. Also, the port used can be changed by passing the --port option.\n"
                     agent-path
                     (string-drop agent-path 1))))))
 
+(define (with-request-mutex thunk)
+  (monitor (thunk)))
+
+(define* (coordinator-handle-failed-request method path response body
+                                            #:key first-request-failed?)
+  (simple-format
+   (current-error-port)
+   "error: coordinator-http-request: ~A ~A: ~A\n"
+   method path (response-code response))
+
+  (catch #t
+    (lambda ()
+      (if (equal? '(application/json (charset . "utf-8"))
+                  (response-content-type response))
+          (json-string->scm (utf8->string body))
+          (utf8->string body)))
+    (lambda (key . args)
+      (simple-format
+       (current-error-port)
+       "error decoding body ~A ~A\n"
+       key args)
+      #f)))
+
 (define* (coordinator-http-request coordinator-uri agent-uuid password
                                    path
                                    #:key method body (headers '())
@@ -464,46 +488,35 @@ port. Also, the port used can be changed by passing the --port option.\n"
 
   (define (make-request)
     (let-values (((response body)
-                  (http-request uri
-                                #:method method
-                                #:body (scm->json-string body)
-                                #:decode-body? #f
-                                #:headers
-                                `((Authorization . ,auth-value)
-                                  ,@headers))))
+                  (with-request-mutex
+                   (lambda ()
+                     (http-request uri
+                                   #:method method
+                                   #:body (scm->json-string body)
+                                   #:decode-body? #f
+                                   #:headers
+                                   `((Authorization . ,auth-value)
+                                     ,@headers))))))
       (if (>= (response-code response) 400)
-          (begin
-            (simple-format
-             (current-error-port)
-             "error: coordinator-http-request: ~A ~A: ~A\n"
-             method path (response-code response))
-            (let ((body
-                   (catch #t
-                     (lambda ()
-                       (if (equal? '(application/json (charset . "utf-8"))
-                                   (response-content-type response))
-                           (json-string->scm (utf8->string body))
-                           (utf8->string body)))
-                     (lambda (key . args)
-                       (simple-format
-                        (current-error-port)
-                        "error decoding body ~A ~A\n"
-                        key args)
-                       #f))))
-              (if (and first-request-failed?
-                       succeed-on-access-denied-retry?
-                       (equal? body
-                               '(("error" . "access denied"))))
-                  (begin
-                    (simple-format
-                     (current-error-port)
-                     "warning: treating access denied response as success\n")
-                    (values body response))
-                  (begin
-                    (set! first-request-failed? #t)
-                    (raise-exception
-                     (make-exception-with-message
-                      body))))))
+          (let ((body
+                 (coordinator-handle-failed-request method
+                                                    path
+                                                    response
+                                                    body)))
+            (if (and first-request-failed?
+                     succeed-on-access-denied-retry?
+                     (equal? body
+                             '(("error" . "access denied"))))
+                (begin
+                  (simple-format
+                   (current-error-port)
+                   "warning: treating access denied response as success\n")
+                  (values body response))
+                (begin
+                  (set! first-request-failed? #t)
+                  (raise-exception
+                   (make-exception-with-message
+                    body)))))
           (values
            (json-string->scm (utf8->string body))
            response))))
@@ -545,14 +558,16 @@ port. Also, the port used can be changed by passing the --port option.\n"
          5000000)                       ; 5MB
       (retry-on-error
        (lambda ()
-         (call-with-streaming-http-request
-          uri
-          (lambda (port)
-            (call-with-lzip-output-port port
-              (lambda (port)
-                (write-file file port))
-              #:level 9))
-          #:headers `((Authorization . ,auth-value))))
+         (with-request-mutex
+          (lambda ()
+            (call-with-streaming-http-request
+             uri
+             (lambda (port)
+               (call-with-lzip-output-port port
+                 (lambda (port)
+                   (write-file file port))
+                 #:level 9))
+             #:headers `((Authorization . ,auth-value))))))
        #:times 3
        #:delay 30)
       (let* ((directory (or (getenv "TMPDIR") "/tmp"))
@@ -570,18 +585,28 @@ port. Also, the port used can be changed by passing the --port option.\n"
         (simple-format #t "finished compressing ~A, now sending\n" file)
         (retry-on-error
          (lambda ()
-           (call-with-input-file template
-             (lambda (file-port)
-               (call-with-streaming-http-request
-                uri
-                (lambda (port)
-                  (with-time-logging
-                      (simple-format #f "sending ~A" file)
-                    (dump-port file-port port
-                               #:buffer-size (expt 2 20))))
-                #:headers `((Authorization . ,auth-value))))))
+           (with-request-mutex
+            (lambda ()
+              (call-with-input-file template
+                (lambda (file-port)
+                  (let-values (((response body)
+                                (call-with-streaming-http-request
+                                 uri
+                                 (lambda (port)
+                                   (with-time-logging
+                                    (simple-format #f "sending ~A" file)
+                                    (dump-port file-port port
+                                               #:buffer-size (expt 2 20))))
+                                 #:headers `((Authorization . ,auth-value)))))
+                    (when (>= (response-code response) 400)
+                      (raise-exception
+                       (make-exception-with-message
+                        (coordinator-handle-failed-request 'PUT
+                                                           (uri-path uri)
+                                                           response
+                                                           body))))))))))
          #:times 9
-         #:delay 30)
+         #:delay (+ 60 (random 120)))
 
         (delete-file template))))
 
@@ -608,16 +633,26 @@ port. Also, the port used can be changed by passing the --port option.\n"
 
   (retry-on-error
    (lambda ()
-     (call-with-streaming-http-request
-      uri
-      (lambda (request-port)
-        (call-with-input-file file
-          (lambda (file-port)
-            (dump-port file-port request-port))
-          #:binary #t))
-      #:headers `((Authorization . ,auth-value))))
+     (with-request-mutex
+      (lambda ()
+        (let-values (((response body)
+                      (call-with-streaming-http-request
+                       uri
+                       (lambda (request-port)
+                         (call-with-input-file file
+                           (lambda (file-port)
+                             (dump-port file-port request-port))
+                           #:binary #t))
+                       #:headers `((Authorization . ,auth-value)))))
+          (when (>= (response-code response) 400)
+            (raise-exception
+             (make-exception-with-message
+              (coordinator-handle-failed-request 'PUT
+                                                 (uri-path uri)
+                                                 response
+                                                 body))))))))
    #:times 9
-   #:delay 30))
+   #:delay (+ 30 (random 60))))
 
 (define (submit-build-result coordinator-uri agent-uuid password
                              build-id result)