Implement retrying for HTTP1.1 as well

HTTP1.1 connection reuse is subject to a race condition where remote
connection tear down races with reusing the connection for a new request.
This change wires HTTP1.1 into the same request retry mechanism HTTP2 uses.

lib/Cro/HTTP/Client.pm6

@@ -119,6 +119,7 @@ class Cro::HTTP::Client::Policy::Timeout does Cro::Policy::Timeout[%(
 #| for multiple requests, as well as allowing configuration of common properties of
 #| many requests at construction time.
 class Cro::HTTP::Client {
+    my class PipelineClosedBeforeHeaders is Exception { }
     my class Pipeline {
         has Bool $.secure;
         has Str $.host;
@@ -127,6 +128,9 @@ class Cro::HTTP::Client {
         has Tap $!tap;
         has $!next-response-vow;
         has Bool $.dead = False;
+        # Locks a race condition with the $!dead attribute.
+        # Has nothing to do with a dead-lock.
+        has Lock::Async $!dead-lock .= new;
 
         submethod BUILD(:$!secure!, :$!host!, :$!port!, :$!in!, :$out!) {
             $!tap = supply {
@@ -135,19 +139,23 @@ class Cro::HTTP::Client {
                     $!next-response-vow = Nil;
                     $vow.keep($_);
                     LAST {
-                        $!dead = True;
-                        if $!next-response-vow {
-                            $!next-response-vow.break:
-                                'Connection unexpectedly closed before response headers received';
-                            $!next-response-vow = Nil;
+                        $!dead-lock.protect: {
+                            $!dead = True;
+                            if $!next-response-vow {
+                                $!next-response-vow.break:
+                                    PipelineClosedBeforeHeaders.new;
+                                $!next-response-vow = Nil;
+                            }
                         }
                     }
                     QUIT {
                         default {
-                            $!dead = True;
-                            if $!next-response-vow {
-                                $!next-response-vow.break($_);
-                                $!next-response-vow = Nil;
+                            $!dead-lock.protect: {
+                                $!dead = True;
+                                if $!next-response-vow {
+                                    $!next-response-vow.break($_);
+                                    $!next-response-vow = Nil;
+                                }
                             }
                         }
                     }
@@ -156,8 +164,21 @@ class Cro::HTTP::Client {
         }
 
         method send-request($request --> Promise) {
-            my $next-response-promise = Promise.new;
-            $!next-response-vow = $next-response-promise.vow;
+            my $next-response-promise;
+            my $broken = False;
+            $!dead-lock.protect: {
+                if $!dead {
+                    # Without https://github.com/MoarVM/MoarVM/pull/1782 merged,
+                    # we can't put the below return here.
+                    $broken = True;
+                }
+                else {
+                    $next-response-promise = Promise.new;
+                    $!next-response-vow = $next-response-promise.vow;
+                }
+            }
+            return Promise.broken(PipelineClosedBeforeHeaders.new) if $broken;
+
             $!in.emit($request);
             return $next-response-promise;
         }
@@ -617,7 +638,6 @@ class Cro::HTTP::Client {
 
                     # Send the request.
                     whenever $pipeline.send-request($request-object) {
-                        $headers-kept = True;
                         QUIT {
                             $request-log.end;
                             when GoAwayRetry {
@@ -628,7 +648,16 @@ class Cro::HTTP::Client {
                                     .goaway-exception.rethrow;
                                 }
                             }
+                            when PipelineClosedBeforeHeaders {
+                                if $goaway-retries > 0 && !$headers-kept {
+                                    $retry-supplier.emit: True;
+                                }
+                                else {
+                                    .rethrow;
+                                }
+                            }
                         }
+                        $headers-kept = True;
 
                         # Consider adding the connection back into the cache to use it
                         # again.