add WebTransport http/2 support to proxygen/lib

Summary:
This diff adds support for WebTransport over http/2 in proxygen/lib for parity with proxygen::coro (recently added). Designed to reuse as much code as possible with proxygen::coro (WtUtils, WtIngressCb, WtEgressCb, etc.)

Diff is admittedly a bit messy, but the abstractions in legacy proxygen/lib aren't as clean as proxygen::coro (pretty much no way around it). Unit tests ported over from proxygen::coro

Reviewed By: joanna-jo, afrind

Differential Revision: D90333392

fbshipit-source-id: 7566379a6b6e74ba7f610398eb105d0bcc9fc4d6

Author: hanidamlaj

proxygen/lib/http/session/CMakeLists.txt

@@ -13,6 +13,7 @@ proxygen_add_library(proxygen_http_session_session
   DEPS
     proxygen_http_codec_http1x_codec
     proxygen_http_session_stats
+    proxygen_http_webtransport_webtransport_session
     fizz::fizz
     Folly::folly_conv
     Folly::folly_cpp_attributes
@@ -184,6 +185,8 @@ proxygen_add_library(proxygen_http_session_http_upstream_session
   SRCS
     HTTPUpstreamSession.cpp
   DEPS
+    proxygen_http_webtransport_httpwebtransport
+    proxygen_http_webtransport_wt_util
     wangle::wangle_acceptor_managed
     Folly::folly_io_async_async_ssl_socket
   EXPORTED_DEPS

proxygen/lib/http/session/HTTPSession.cpp

@@ -21,6 +21,8 @@
 #include <proxygen/lib/http/codec/HTTPChecks.h>
 #include <proxygen/lib/http/session/HTTPSessionController.h>
 #include <proxygen/lib/http/session/HTTPSessionStats.h>
+#include <proxygen/lib/http/webtransport/WebTransportSession.h>
+#include <proxygen/lib/http/webtransport/WtUtils.h>
 #include <wangle/acceptor/ConnectionManager.h>
 
 using fizz::AsyncFizzBase;
@@ -46,6 +48,7 @@ static constexpr folly::StringPiece kClientLabel =
     "EXPORTER HTTP CERTIFICATE client";
 static constexpr folly::StringPiece kServerLabel =
     "EXPORTER HTTP CERTIFICATE server";
+
 } // anonymous namespace
 
 namespace proxygen {
@@ -275,6 +278,8 @@ std::chrono::milliseconds HTTPSession::getDrainTimeout() const {
 void HTTPSession::startNow() {
   CHECK(!started_);
   started_ = true;
+  detail::setEgressWtHttpSettings(codec_->getEgressSettings());
+
   codec_->generateSettings(writeBuf_);
   if (connFlowControl_) {
     connFlowControl_->setReceiveWindowSize(writeBuf_,
@@ -1449,6 +1454,44 @@ void HTTPSession::sendHeaders(HTTPTransaction* txn,
           observer->requestStarted(observed, event);
         });
   }
+  // terminate WebTransport if connect stream, wtHandler_ set and upstream or
+  // 2xx resp
+  auto& wtCtx = txn->wtCtx_;
+  const bool upgraded = txn->isWebTransportConnectStream() &&
+                        (headers.isRequest() || headers.is2xxResponse());
+  const bool makeWtSession = !includeEOM && upgraded && wtCtx.hasWtHandler();
+  VLOG(6) << "eom=" << includeEOM << "; upgraded=" << upgraded
+          << "; wtConnect=" << txn->isWebTransportConnectStream()
+          << "; supportsWebTransport=" << supportsWebTransport()
+          << "; wtHandler=" << wtCtx.hasWtHandler();
+  if (!makeWtSession) {
+    return;
+  }
+  /**
+   * for downstream:
+   *     Detach existing txn handler with a custom WtHandler to transparently
+   *     enable wt.
+   *
+   * for upstream:
+   *     Implementation detail, but all wt requests should go thru
+   *     HTTPUpstreamSession::sendWtReq. The existing handler still needs to be
+   *     notified of ::onHeadersComplete or ::onError (whichever happens first)
+   *     to resolve the Promise/Future returned to the application.
+   */
+  if (auto* prevHandler = txn->getHandler()) {
+    prevHandler->detachTransaction();
+  }
+  txn->setHandler(nullptr); // clear existing handler
+  // make wt session
+  detail::WtDir dir =
+      isUpstream() ? detail::WtDir::Client : detail::WtDir::Server;
+  auto wtConfig = detail::getWtConfig(codec_->getIngressSettings(),
+                                      codec_->getEgressSettings());
+  auto wtSession = detail::H2WtSession::make(
+      getEventBase(), dir, wtConfig, wtCtx.moveWtHandler());
+  // ::init notifies WtHandler of WtSession
+  wtSession->init(wtSession, isUpstream() ? wtCtx.moveWtCallback() : nullptr);
+  txn->setHandler(&wtSession->getTxnHandler());
 }
 
 void HTTPSession::commonEom(HTTPTransaction* txn,
@@ -2946,4 +2989,10 @@ void HTTPSession::invokeOnAllTransactions(
   }
 }
 
+bool HTTPSession::supportsWebTransport() const noexcept {
+  const auto& codec = getCodec();
+  return detail::supportsH2Wt(
+      {codec.getIngressSettings(), codec.getEgressSettings()});
+}
+
 } // namespace proxygen

proxygen/lib/http/session/HTTPSession.h

@@ -501,6 +501,7 @@ class HTTPSession
   const HTTPCodec& getCodec() const noexcept override {
     return codec_.getChainEnd();
   }
+  bool supportsWebTransport() const noexcept final;
 
   /**
    * Returns the underlying AsyncTransport.

proxygen/lib/http/session/HTTPTransaction.cpp

@@ -47,6 +47,11 @@ inline ErrorCode getDefaultAbortErrorCode(bool isUpstream) {
   return isUpstream ? ErrorCode::CANCEL : ErrorCode::INTERNAL_ERROR;
 }
 
+bool isConnectUdp(const HTTPMessage& msg) noexcept {
+  return msg.getMethod() == HTTPMethod::CONNECT && msg.getUpgradeProtocol() &&
+         *msg.getUpgradeProtocol() == "connect-udp";
+}
+
 } // namespace
 
 #define INVARIANT_RETURN(X, Y)                                            \
@@ -217,9 +222,7 @@ void HTTPTransaction::onIngressHeadersComplete(
     headRequest_ = (method == HTTPMethod::HEAD);
     upgraded_ = (method == HTTPMethod::CONNECT);
     wtConnectStream_ = HTTPWebTransport::isConnectMessage(*msg);
-    connectUdpStream_ = msg->getMethod() == HTTPMethod::CONNECT &&
-                        msg->getUpgradeProtocol() &&
-                        *msg->getUpgradeProtocol() == "connect-udp";
+    connectUdpStream_ = isConnectUdp(*msg);
   }
 
   if ((msg->isRequest() && msg->getMethod() != HTTPMethod::CONNECT) ||
@@ -1026,12 +1029,11 @@ void HTTPTransaction::sendHeadersWithOptionalEOM(const HTTPMessage& headers,
   if (!headers.isRequest() && !isPushed()) {
     lastResponseStatus_ = headers.getStatusCode();
   }
+
   if (headers.isRequest()) {
     headRequest_ = (headers.getMethod() == HTTPMethod::HEAD);
     wtConnectStream_ = HTTPWebTransport::isConnectMessage(headers);
-    connectUdpStream_ = headers.getMethod() == HTTPMethod::CONNECT &&
-                        headers.getUpgradeProtocol() &&
-                        *headers.getUpgradeProtocol() == "connect-udp";
+    connectUdpStream_ = isConnectUdp(headers);
   } else {
     has1xxResponse_ = headers.is1xxResponse();
   }
@@ -1101,6 +1103,16 @@ void HTTPTransaction::sendHeaders(const HTTPMessage& header) {
   sendHeadersWithOptionalEOM(header, false);
 }
 
+void HTTPTransaction::sendWtHeaders(
+    const HTTPMessage& headers,
+    WebTransportHandler::Ptr wtHandler,
+    HttpWtClientCallbackPtr wtClientCb) noexcept {
+  wtCtx_.wtHandler_ = std::move(wtHandler);
+  wtCtx_.upstreamWtCb_ = std::move(wtClientCb);
+  CHECK(wtCtx_.wtHandler_);
+  sendHeadersWithOptionalEOM(headers, /*eom=*/false);
+}
+
 void HTTPTransaction::sendBody(std::unique_ptr<folly::IOBuf> body) {
   DestructorGuard guard(this);
   bool chunking =

proxygen/lib/http/session/HTTPTransaction.h

@@ -430,6 +430,9 @@ class HTTPTransactionTransportCallback {
 };
 
 class HTTPSessionBase;
+struct HttpWtClientCallbackIf;
+using HttpWtClientCallbackPtr = std::unique_ptr<HttpWtClientCallbackIf>;
+
 class HTTPTransaction
     : public folly::HHWheelTimer::Callback
     , public folly::DelayedDestructionBase
@@ -1170,6 +1173,26 @@ class HTTPTransaction
   virtual void sendHeadersWithEOM(const HTTPMessage& headers);
   virtual void sendHeadersWithOptionalEOM(const HTTPMessage& headers, bool eom);
 
+  /**
+   * Variant of send headers specifically for webtransport requests/responses
+   *
+   * This api will terminate WebTransport. If 200/ok resp is
+   * sent, the existing application's HTTPTransactionHandler will be detached
+   * and replaced internally. Applications should solely interact with the
+   * WebTransport session via the WebTransportHandler & WebTransportSession
+   * (passed into WebTransportHandler::onWebTransportSession) pair.
+   *
+   * If an endpoint/proxy wants to bypass parsing WebTransport bits (e.g. to
+   * enable http/2 end-to-end CONNECT tunneling), it can use the non-wt
+   * HttpTxn::sendHeaders
+   *
+   * HttpWtCallbackPtr is solely for upstream -- it's a mechanism to be notified
+   * of ingress headers/error (whichever occurs first).
+   */
+  void sendWtHeaders(const HTTPMessage& headers,
+                     WebTransportHandler::Ptr wtHandler,
+                     HttpWtClientCallbackPtr wtClientCb = nullptr) noexcept;
+
   /**
    * Send part or all of the egress message body to the Transport. If flow
    * control is enabled, the chunk boundaries may not be respected.
@@ -1576,7 +1599,7 @@ class HTTPTransaction
 
   folly::Optional<ConnectionToken> getConnectionToken() const noexcept;
 
-  bool isWebTransportConnectStream() {
+  bool isWebTransportConnectStream() const {
     return transport_.supportsWebTransport() && wtConnectStream_;
   }
 
@@ -1634,6 +1657,29 @@ class HTTPTransaction
     return &txnObserverAccessor_;
   }
 
+  // only for consumption by the backing sessions (e.g. HTTPSession, HQSession)
+  // avoids applications messing with wt state
+  struct WtCtx {
+    friend class HTTPTransaction;
+    friend class HTTPSession;
+    friend class HQSession;
+
+   private:
+    bool hasWtHandler() const noexcept {
+      return bool(wtHandler_);
+    }
+    WebTransportHandler::Ptr moveWtHandler() noexcept {
+      return std::move(wtHandler_);
+    }
+
+    HttpWtClientCallbackPtr moveWtCallback() noexcept {
+      return std::move(upstreamWtCb_);
+    }
+
+    WebTransportHandler::Ptr wtHandler_;
+    HttpWtClientCallbackPtr upstreamWtCb_;
+  } wtCtx_;
+
  private:
   HTTPTransaction(const HTTPTransaction&) = delete;
   HTTPTransaction& operator=(const HTTPTransaction&) = delete;
@@ -2013,6 +2059,14 @@ class HTTPTransaction
   HTTPTransactionObserverContainer txnObserverContainer_;
 };
 
+struct HttpWtClientCallbackIf {
+  virtual ~HttpWtClientCallbackIf() noexcept = default;
+  virtual void onHeaders(std::unique_ptr<HTTPMessage>) noexcept {
+  }
+  virtual void onErr(const HTTPException&) noexcept {
+  }
+};
+
 /**
  * Write a description of an HTTPTransaction to an ostream
  */

proxygen/lib/http/session/HTTPUpstreamSession.cpp

@@ -8,6 +8,9 @@
 
 #include <proxygen/lib/http/session/HTTPUpstreamSession.h>
 
+#include <proxygen/lib/http/webtransport/HTTPWebTransport.h>
+#include <proxygen/lib/http/webtransport/WtUtils.h>
+
 #include <folly/io/async/AsyncSSLSocket.h>
 #include <wangle/acceptor/ConnectionManager.h>
 
@@ -103,10 +106,7 @@ void HTTPUpstreamSession::startNow() {
 HTTPTransaction* HTTPUpstreamSession::newTransaction(
     HTTPTransaction::Handler* handler) {
   auto txn = newTransactionWithError(handler);
-  if (txn.hasError()) {
-    return nullptr;
-  }
-  return txn.value();
+  return txn.value_or(nullptr);
 }
 
 folly::Expected<HTTPTransaction*, HTTPUpstreamSession::NewTransactionError>
@@ -224,6 +224,109 @@ void HTTPUpstreamSession::detachThreadLocals(bool detachSSLContext) {
   }
 }
 
+namespace {
+
+constexpr std::string_view kWtNotSupported = "WebTransport not supported";
+constexpr std::string_view kInvalidWtReq = "Invalid WebTransport request";
+constexpr std::string_view kStreamFailed = "Failed to create stream";
+
+folly::exception_wrapper makeHttpEx(const std::string& err) noexcept {
+  constexpr auto kExDir = HTTPException::Direction::INGRESS_AND_EGRESS;
+  return folly::make_exception_wrapper<HTTPException>(kExDir, err);
+}
+
+using WtReqResult = std::unique_ptr<HTTPMessage>;
+using WtReqResultPromise = folly::Promise<WtReqResult>;
+folly::Promise<WtReqResult> emptyWtReqPromise() noexcept {
+  return folly::Promise<WtReqResult>::makeEmpty();
+}
+
+class WtClientCallback final
+    : public HttpWtClientCallbackIf
+    , public HTTPTransactionHandler {
+ private:
+  folly::Promise<WtReqResult> promise{emptyWtReqPromise()};
+
+ public:
+  explicit WtClientCallback(WtReqResultPromise p) noexcept
+      : promise(std::move(p)) {
+  }
+  ~WtClientCallback() noexcept override = default;
+  folly::Promise<WtReqResult> resetPromise() noexcept {
+    return std::exchange(promise, emptyWtReqPromise());
+  }
+  /**
+   * Either ::onHeaders or ::onErr can be invoked first
+   *
+   * When ::onHeadersComplete is invoked, resolve the promise with the non-final
+   * http headers
+   *
+   * When ::onError is invoked, resolve the promise with HTTPException
+   */
+  void onHeaders(std::unique_ptr<HTTPMessage> msg) noexcept override {
+    if (msg->isFinal()) {
+      auto p = resetPromise();
+      CHECK(p.valid());
+      p.setValue(std::move(msg));
+    }
+  }
+  void onErr(const HTTPException& ex) noexcept override {
+    auto p = resetPromise();
+    CHECK(p.valid());
+    p.setException(ex);
+  }
+
+  // **ignored**
+  void onHeadersComplete(std::unique_ptr<HTTPMessage>) noexcept override {
+  }
+  void onError(const HTTPException&) noexcept override {
+  }
+  void detachTransaction() noexcept override {
+  }
+  void setTransaction(HTTPTransaction*) noexcept override {
+  }
+  void onBody(std::unique_ptr<folly::IOBuf>) noexcept override {
+  }
+  void onTrailers(std::unique_ptr<HTTPHeaders>) noexcept override {
+  }
+  void onEOM() noexcept override {
+  }
+  void onUpgrade(UpgradeProtocol) noexcept override {
+  }
+  void onEgressPaused() noexcept override {
+  }
+  void onEgressResumed() noexcept override {
+  }
+};
+
+} // namespace
+
+folly::SemiFuture<WtReqResult> HTTPUpstreamSession::sendWebTransportRequest(
+    const HTTPMessage& req, WebTransportHandler::Ptr wtHandler) noexcept {
+  // both self and peer must indicate support for WebTransport
+  const bool supportsWt = proxygen::detail::supportsH2Wt(
+      {codec_->getIngressSettings(), codec_->getEgressSettings()});
+  const bool validWtReq = HTTPWebTransport::isConnectMessage(req);
+  if (!(supportsWt && validWtReq)) {
+    auto err = !validWtReq ? kInvalidWtReq : kWtNotSupported;
+    VLOG(6) << __func__ << " err=" << err << "; sess=" << *this;
+    return makeHttpEx(std::string(err));
+  }
+
+  auto [p, f] = folly::makePromiseContract<WtReqResult>();
+  auto wtClientCb = std::make_unique<WtClientCallback>(std::move(p));
+
+  auto* txn = newTransaction(wtClientCb.get());
+  if (!txn) {
+    return makeHttpEx(std::string(kStreamFailed));
+  }
+
+  // send wt upgrade req
+  txn->setHandler(nullptr); // clear handler, will be replaced by sendWtHeaders
+  txn->sendWtHeaders(req, std::move(wtHandler), std::move(wtClientCb));
+  return std::move(f);
+}
+
 void HTTPUpstreamSession::maybeDetachSSLContext() const {
 #ifndef NO_ASYNCSSLSOCKET
   auto sslSocket = sock_->getUnderlyingTransport<folly::AsyncSSLSocket>();