Make HttpClient adapter preserve exception behavior

Ideally, the HttpClient/HttpServer adapters should maintain the invariant that
the behavior of a given client is the same as the behavior of
newHttpClient(newHttpService(client)).  Prior to this change, the HttpClient
wrapper lazily called request(), so a client whose request() eagerly threw an
exception could produce a different exception when called directly versus when
wrapped -- the laziness allowed additional code to run.  This was particularly
evident when making a request with a body, since code using a wrapped client
would be able to set up a subsequent BlockedWrite, eventually resulting in a
"read end of pipe was aborted" exception instead of the actual exception.

This change makes HttpClientAdapter::request() execute the wrapped request()
eagerly.

(Note that it partially undoes 90d48343... Hopefully, it preserves the desired
behavior added there.)

c++/src/kj/compat/http-test.c++

@@ -3032,6 +3032,39 @@ KJ_TEST("newHttpClient from HttpService WebSockets") {
   testWebSocketClient(waitScope, *headerTable, hMyHeader, *client);
 }
 
+KJ_TEST("adapted client/server propagates request exceptions like non-adapted client") {
+  auto io = kj::setupAsyncIo();
+
+  HttpHeaderTable table;
+  HttpHeaders headers(table);
+
+  class FailingHttpClient final: public HttpClient {
+  public:
+    Request request(
+        HttpMethod method, kj::StringPtr url, const HttpHeaders& headers,
+        kj::Maybe<uint64_t> expectedBodySize = nullptr) override {
+      KJ_FAIL_ASSERT("request_fail");
+    }
+
+    kj::Promise<WebSocketResponse> openWebSocket(
+        kj::StringPtr url, const HttpHeaders& headers) override {
+      KJ_FAIL_ASSERT("websocket_fail");
+    }
+  };
+
+  auto rawClient = kj::heap<FailingHttpClient>();
+
+  auto innerClient = kj::heap<FailingHttpClient>();
+  auto adaptedService = kj::newHttpService(*innerClient).attach(kj::mv(innerClient));
+  auto adaptedClient = kj::newHttpClient(*adaptedService).attach(kj::mv(adaptedService));
+
+  KJ_EXPECT_THROW_MESSAGE("request_fail", rawClient->request(HttpMethod::POST, "/"_kj, headers));
+  KJ_EXPECT_THROW_MESSAGE("request_fail", adaptedClient->request(HttpMethod::POST, "/"_kj, headers));
+
+  KJ_EXPECT_THROW_MESSAGE("websocket_fail", rawClient->openWebSocket("/"_kj, headers));
+  KJ_EXPECT_THROW_MESSAGE("websocket_fail", adaptedClient->openWebSocket("/"_kj, headers));
+}
+
 // -----------------------------------------------------------------------------
 
 class CountingIoStream final: public kj::AsyncIoStream {

c++/src/kj/compat/http.c++

@@ -4294,15 +4294,13 @@ public:
     //   involving a PromiseAdapter, maybe?
     auto paf = kj::newPromiseAndFulfiller<Response>();
     auto responder = kj::refcounted<ResponseImpl>(method, kj::mv(paf.fulfiller));
-    auto promise = kj::evalLater([this, method,
-                                  urlCopy = kj::mv(urlCopy),
-                                  headersCopy = kj::mv(headersCopy),
-                                  pipeIn = kj::mv(pipe.in),
-                                  &responder = *responder]() mutable {
-      auto promise = service.request(method, urlCopy, *headersCopy, *pipeIn, responder);
-      return promise.attach(kj::mv(pipeIn), kj::mv(urlCopy), kj::mv(headersCopy));
-    });
-    responder->setPromise(kj::mv(promise));
+
+    auto requestPaf = kj::newPromiseAndFulfiller<kj::Promise<void>>();
+    responder->setPromise(kj::mv(requestPaf.promise));
+
+    auto promise = service.request(method, urlCopy, *headersCopy, *pipe.in, *responder)
+        .attach(kj::mv(pipe.in), kj::mv(urlCopy), kj::mv(headersCopy));
+    requestPaf.fulfiller->fulfill(kj::mv(promise));
 
     return {
       kj::mv(pipe.out),
@@ -4323,15 +4321,14 @@ public:
 
     auto paf = kj::newPromiseAndFulfiller<WebSocketResponse>();
     auto responder = kj::refcounted<WebSocketResponseImpl>(kj::mv(paf.fulfiller));
-    auto promise = kj::evalLater([this,
-                                  urlCopy = kj::mv(urlCopy),
-                                  headersCopy = kj::mv(headersCopy),
-                                  &responder = *responder]() mutable {
-      auto in = kj::heap<NullInputStream>();
-      auto promise = service.request(HttpMethod::GET, urlCopy, *headersCopy, *in, responder);
-      return promise.attach(kj::mv(in), kj::mv(urlCopy), kj::mv(headersCopy));
-    });
-    responder->setPromise(kj::mv(promise));
+
+    auto requestPaf = kj::newPromiseAndFulfiller<kj::Promise<void>>();
+    responder->setPromise(kj::mv(requestPaf.promise));
+
+    auto in = kj::heap<NullInputStream>();
+    auto promise = service.request(HttpMethod::GET, urlCopy, *headersCopy, *in, *responder)
+        .attach(kj::mv(in), kj::mv(urlCopy), kj::mv(headersCopy));
+    requestPaf.fulfiller->fulfill(kj::mv(promise));
 
     return paf.promise.attach(kj::mv(responder));
   }