Handle HttpService failing to call response.send() or throwing disconnect.

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

@@ -345,8 +345,12 @@ kj::Promise<void> expectRead(kj::AsyncInputStream& in, kj::StringPtr expected) {
 
   auto buffer = kj::heapArray<char>(expected.size());
 
-  auto promise = in.read(buffer.begin(), 1, buffer.size());
+  auto promise = in.tryRead(buffer.begin(), 1, buffer.size());
   return promise.then(kj::mvCapture(buffer, [&in,expected](kj::Array<char> buffer, size_t amount) {
+    if (amount == 0) {
+      KJ_FAIL_ASSERT("expected data never sent", expected);
+    }
+
     auto actual = buffer.slice(0, amount);
     if (memcmp(actual.begin(), expected.begin(), actual.size()) != 0) {
       KJ_FAIL_ASSERT("data from stream doesn't match expected", expected, actual);
@@ -1102,6 +1106,170 @@ KJ_TEST("HttpServer pipeline timeout") {
   KJ_EXPECT(pipe.ends[1]->readAllText().wait(io.waitScope) == "");
 }
 
+class BrokenHttpService final: public HttpService {
+  // HttpService that doesn't send a response.
+public:
+  BrokenHttpService() = default;
+  explicit BrokenHttpService(kj::Exception&& exception): exception(kj::mv(exception)) {}
+
+  kj::Promise<void> request(
+      HttpMethod method, kj::StringPtr url, const HttpHeaders& headers,
+      kj::AsyncInputStream& requestBody, Response& responseSender) override {
+    return requestBody.readAllBytes().then([this](kj::Array<byte>&&) -> kj::Promise<void> {
+      KJ_IF_MAYBE(e, exception) {
+        return kj::cp(*e);
+      } else {
+        return kj::READY_NOW;
+      }
+    });
+  }
+
+private:
+  kj::Maybe<kj::Exception> exception;
+};
+
+KJ_TEST("HttpServer no response") {
+  auto io = kj::setupAsyncIo();
+  auto pipe = io.provider->newTwoWayPipe();
+
+  HttpHeaderTable table;
+  BrokenHttpService service;
+  HttpServer server(io.provider->getTimer(), table, service);
+
+  auto listenTask = server.listenHttp(kj::mv(pipe.ends[0]));
+
+  // Do one request.
+  pipe.ends[1]->write(PIPELINE_TESTS[0].request.raw.begin(), PIPELINE_TESTS[0].request.raw.size())
+      .wait(io.waitScope);
+  auto text = pipe.ends[1]->readAllText().wait(io.waitScope);
+
+  KJ_EXPECT(text ==
+      "HTTP/1.1 500 Internal Server Error\r\n"
+      "Connection: close\r\n"
+      "Content-Length: 51\r\n"
+      "Content-Type: text/plain\r\n"
+      "\r\n"
+      "ERROR: The HttpService did not generate a response.", text);
+}
+
+KJ_TEST("HttpServer disconnected") {
+  auto io = kj::setupAsyncIo();
+  auto pipe = io.provider->newTwoWayPipe();
+
+  HttpHeaderTable table;
+  BrokenHttpService service(KJ_EXCEPTION(DISCONNECTED, "disconnected"));
+  HttpServer server(io.provider->getTimer(), table, service);
+
+  auto listenTask = server.listenHttp(kj::mv(pipe.ends[0]));
+
+  // Do one request.
+  pipe.ends[1]->write(PIPELINE_TESTS[0].request.raw.begin(), PIPELINE_TESTS[0].request.raw.size())
+      .wait(io.waitScope);
+  auto text = pipe.ends[1]->readAllText().wait(io.waitScope);
+
+  KJ_EXPECT(text == "", text);
+}
+
+KJ_TEST("HttpServer overloaded") {
+  auto io = kj::setupAsyncIo();
+  auto pipe = io.provider->newTwoWayPipe();
+
+  HttpHeaderTable table;
+  BrokenHttpService service(KJ_EXCEPTION(OVERLOADED, "overloaded"));
+  HttpServer server(io.provider->getTimer(), table, service);
+
+  auto listenTask = server.listenHttp(kj::mv(pipe.ends[0]));
+
+  // Do one request.
+  pipe.ends[1]->write(PIPELINE_TESTS[0].request.raw.begin(), PIPELINE_TESTS[0].request.raw.size())
+      .wait(io.waitScope);
+  auto text = pipe.ends[1]->readAllText().wait(io.waitScope);
+
+  KJ_EXPECT(text.startsWith("HTTP/1.1 503 Service Unavailable"), text);
+}
+
+KJ_TEST("HttpServer unimplemented") {
+  auto io = kj::setupAsyncIo();
+  auto pipe = io.provider->newTwoWayPipe();
+
+  HttpHeaderTable table;
+  BrokenHttpService service(KJ_EXCEPTION(UNIMPLEMENTED, "unimplemented"));
+  HttpServer server(io.provider->getTimer(), table, service);
+
+  auto listenTask = server.listenHttp(kj::mv(pipe.ends[0]));
+
+  // Do one request.
+  pipe.ends[1]->write(PIPELINE_TESTS[0].request.raw.begin(), PIPELINE_TESTS[0].request.raw.size())
+      .wait(io.waitScope);
+  auto text = pipe.ends[1]->readAllText().wait(io.waitScope);
+
+  KJ_EXPECT(text.startsWith("HTTP/1.1 501 Not Implemented"), text);
+}
+
+KJ_TEST("HttpServer threw exception") {
+  auto io = kj::setupAsyncIo();
+  auto pipe = io.provider->newTwoWayPipe();
+
+  HttpHeaderTable table;
+  BrokenHttpService service(KJ_EXCEPTION(FAILED, "failed"));
+  HttpServer server(io.provider->getTimer(), table, service);
+
+  auto listenTask = server.listenHttp(kj::mv(pipe.ends[0]));
+
+  // Do one request.
+  pipe.ends[1]->write(PIPELINE_TESTS[0].request.raw.begin(), PIPELINE_TESTS[0].request.raw.size())
+      .wait(io.waitScope);
+  auto text = pipe.ends[1]->readAllText().wait(io.waitScope);
+
+  KJ_EXPECT(text.startsWith("HTTP/1.1 500 Internal Server Error"), text);
+}
+
+class PartialResponseService final: public HttpService {
+  // HttpService that sends a partial response then throws.
+public:
+  kj::Promise<void> request(
+      HttpMethod method, kj::StringPtr url, const HttpHeaders& headers,
+      kj::AsyncInputStream& requestBody, Response& response) override {
+    return requestBody.readAllBytes()
+        .then([this,&response](kj::Array<byte>&&) -> kj::Promise<void> {
+      HttpHeaders headers(table);
+      auto body = response.send(200, "OK", headers, 32);
+      auto promise = body->write("foo", 3);
+      return promise.attach(kj::mv(body)).then([]() -> kj::Promise<void> {
+        return KJ_EXCEPTION(FAILED, "failed");
+      });
+    });
+  }
+
+private:
+  kj::Maybe<kj::Exception> exception;
+  HttpHeaderTable table;
+};
+
+KJ_TEST("HttpServer threw exception after starting response") {
+  auto io = kj::setupAsyncIo();
+  auto pipe = io.provider->newTwoWayPipe();
+
+  HttpHeaderTable table;
+  PartialResponseService service;
+  HttpServer server(io.provider->getTimer(), table, service);
+
+  auto listenTask = server.listenHttp(kj::mv(pipe.ends[0]));
+
+  KJ_EXPECT_LOG(ERROR, "HttpService threw exception after generating a partial response");
+
+  // Do one request.
+  pipe.ends[1]->write(PIPELINE_TESTS[0].request.raw.begin(), PIPELINE_TESTS[0].request.raw.size())
+      .wait(io.waitScope);
+  auto text = pipe.ends[1]->readAllText().wait(io.waitScope);
+
+  KJ_EXPECT(text ==
+      "HTTP/1.1 200 OK\r\n"
+      "Content-Length: 32\r\n"
+      "\r\n"
+      "foo", text);
+}
+
 // -----------------------------------------------------------------------------
 
 KJ_TEST("HttpClient to capnproto.org") {

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

@@ -736,15 +736,29 @@ public:
     // Waits until more data is available, but doesn't consume it. Only meant for server-side use,
     // after a request is handled, to check for pipelined requests. Returns false on EOF.
 
-    if (leftover != nullptr) {
+    // Slightly-crappy code to snarf the expected line break. This will actually eat the leading
+    // regex /\r*\n?/.
+    while (lineBreakBeforeNextHeader && leftover.size() > 0) {
+      if (leftover[0] == '\r') {
+        leftover = leftover.slice(1, leftover.size());
+      } else if (leftover[0] == '\n') {
+        leftover = leftover.slice(1, leftover.size());
+        lineBreakBeforeNextHeader = false;
+      } else {
+        // Err, missing line break, whatever.
+        lineBreakBeforeNextHeader = false;
+      }
+    }
+
+    if (!lineBreakBeforeNextHeader && leftover != nullptr) {
       return true;
     }
 
     return inner.tryRead(headerBuffer.begin(), 1, headerBuffer.size())
-        .then([this](size_t amount) {
+        .then([this](size_t amount) -> kj::Promise<bool> {
       if (amount > 0) {
         leftover = headerBuffer.slice(0, amount);
-        return true;
+        return awaitNextMessage();
       } else {
         return false;
       }
@@ -1616,6 +1630,11 @@ public:
         auto body = httpInput.getEntityBody(
             HttpInputStream::REQUEST, req->method, 0, req->connectionHeaders);
 
+        // TODO(perf): If the client disconnects, should we cancel the response? Probably, to
+        //   prevent permanent deadlock. It's slightly weird in that arguably the client should
+        //   be able to shutdown the upstream but still wait on the downstream, but I believe many
+        //   other HTTP servers do similar things.
+
         auto promise = server.service.request(
             req->method, req->url, httpInput.getHeaders(), *body, *this);
         return promise.attach(kj::mv(body))
@@ -1623,6 +1642,11 @@ public:
             .then([this]() -> kj::Promise<void> {
           // Response done. Await next request.
 
+          if (currentMethod != nullptr) {
+            return sendError(500, "Internal Server Error", kj::str(
+                "ERROR: The HttpService did not generate a response."));
+          }
+
           if (server.draining) {
             // Never mind, drain time.
             return kj::READY_NOW;
@@ -1649,12 +1673,29 @@ public:
         return sendError(400, "Bad Request", kj::str(
             "ERROR: The headers sent by your client were not valid."));
       }
-    }).catch_([this](kj::Exception&& e) {
+    }).catch_([this](kj::Exception&& e) -> kj::Promise<void> {
       // Exception; report 500.
 
+      if (currentMethod == nullptr) {
+        // Dang, already sent a partial response. Can't do anything else.
+        KJ_LOG(ERROR, "HttpService threw exception after generating a partial response",
+                      "too late to report error to client", e);
+        return kj::READY_NOW;
+      }
+
       if (e.getType() == kj::Exception::Type::OVERLOADED) {
         return sendError(503, "Service Unavailable", kj::str(
             "ERROR: The server is temporarily unable to handle your request. Details:\n\n", e));
+      } else if (e.getType() == kj::Exception::Type::UNIMPLEMENTED) {
+        return sendError(501, "Not Implemented", kj::str(
+            "ERROR: The server does not implement this operation. Details:\n\n", e));
+      } else if (e.getType() == kj::Exception::Type::DISCONNECTED) {
+        // How do we tell an HTTP client that there was a transient network error, and it should
+        // try again immediately? There's no HTTP status code for this (503 is meant for "try
+        // again later, not now"). Here's an idea: Don't send any response; just close the
+        // connection, so that it looks like the connection between the HTTP client and server
+        // was dropped. A good client should treat this exactly the way we want.
+        return kj::READY_NOW;
       } else {
         return sendError(500, "Internal Server Error", kj::str(
             "ERROR: The server threw an exception. Details:\n\n", e));