Share the h3 wt CONNECT-stream capsule callback

Summary:
HqWtSession.cpp had a file-local callback that turned capsules from the http/3
CONNECT stream into H3WtSession calls. The coro http/3 WebTransport session
needs the same translation, so rather than write a second copy this moves it
next to H3WtSession as `H3WtCapsuleCallback`. Two things change while moving
it: `onDatagram` now hands the datagram to the session instead of dropping it,
and `onConnectionError` closes the session with a clearer message.

Reviewed By: hanidamlaj

Differential Revision: D116105652

fbshipit-source-id: d590a3940e82b39afa2fe077ecc2c37a3374dd65

Author: afrind

proxygen/lib/http/webtransport/HqWtSession.cpp

@@ -8,66 +8,6 @@
 
 #include <proxygen/lib/http/webtransport/HqWtSession.h>
 
-namespace {
-using namespace proxygen;
-using namespace proxygen::detail;
-
-struct H3CapsuleCodecCb final : public WebTransportCapsuleCodec::Callback {
-  H3WtSession& wtSess;
-  explicit H3CapsuleCodecCb(H3WtSession& wtSess) : wtSess(wtSess) {
-  }
-
-  void onMaxData(WTMaxDataCapsule c) noexcept override {
-    VLOG(4) << __func__;
-    wtSess.onConnMaxData({.maxData = c.maximumData});
-  }
-  void onMaxStreamsBidi(WTMaxStreamsCapsule c) noexcept override {
-    VLOG(4) << __func__;
-    wtSess.onMaxStreams(WtStreamManager::MaxStreamsBidi{c.maximumStreams});
-  }
-  void onMaxStreamsUni(WTMaxStreamsCapsule c) noexcept override {
-    VLOG(4) << __func__;
-    wtSess.onMaxStreams(WtStreamManager::MaxStreamsUni{c.maximumStreams});
-  }
-  void onDrainSession(DrainWebTransportSessionCapsule) noexcept override {
-    VLOG(4) << __func__;
-    wtSess.onDrainSession({});
-  }
-  void onCloseSession(CloseWebTransportSessionCapsule c) noexcept override {
-    VLOG(4) << __func__;
-    wtSess.onCloseSession(WtStreamManager::CloseSession{
-        c.applicationErrorCode, std::move(c.applicationErrorMessage)});
-  }
-  void onConnectionError(CapsuleCodec::ErrorCode error) noexcept override {
-    VLOG(4) << __func__;
-    onCloseSession({uint8_t(error), "onConnectionError"});
-  }
-
-  // ignored callbacks
-  void onMaxStreamData(WTMaxStreamDataCapsule) noexcept override {
-  }
-  void onResetStream(WTResetStreamCapsule) noexcept override {
-  }
-  void onStopSending(WTStopSendingCapsule) noexcept override {
-  }
-  void onStream(WTStreamCapsule) noexcept override {
-  }
-  void onStreamDataBlocked(WTStreamDataBlockedCapsule) noexcept override {
-  }
-  void onStreamsBlockedBidi(WTStreamsBlockedCapsule) noexcept override {
-  }
-  void onStreamsBlockedUni(WTStreamsBlockedCapsule) noexcept override {
-  }
-  void onPadding(PaddingCapsule) noexcept override {
-  }
-  void onDatagram(DatagramCapsule) noexcept override {
-  }
-  void onDataBlocked(WTDataBlockedCapsule) noexcept override {
-  }
-};
-
-}; // namespace
-
 namespace proxygen::detail {
 
 struct WtReadLooper : public WtLooper {
@@ -86,7 +26,7 @@ struct WtReadLooper : public WtLooper {
   void runLoopCallback() noexcept override;
   HqWtSession& wtSess_;
   WebTransportTxnHandler& wtTxnHandler_;
-  H3CapsuleCodecCb codecCb_;
+  H3WtCapsuleCallback codecCb_;
   WebTransportCapsuleCodec codec_;
 };
 

proxygen/lib/http/webtransport/QuicWtSession.cpp

@@ -590,4 +590,49 @@ void H3ConnectStreamCallback::onEvent(WtStreamManager::Event&& ev) noexcept {
   std::visit(visitor, ev);
 }
 
+H3WtCapsuleCallback::H3WtCapsuleCallback(H3WtSession& session) noexcept
+    : h3Wt_(session) {
+}
+
+void H3WtCapsuleCallback::onMaxData(WTMaxDataCapsule c) noexcept {
+  XLOG(DBG6) << __func__ << "; maxData=" << c.maximumData;
+  h3Wt_.onConnMaxData({.maxData = c.maximumData});
+}
+
+void H3WtCapsuleCallback::onMaxStreamsBidi(WTMaxStreamsCapsule c) noexcept {
+  XLOG(DBG6) << __func__ << "; maxStreams=" << c.maximumStreams;
+  h3Wt_.onMaxStreams(WtStreamManager::MaxStreamsBidi{c.maximumStreams});
+}
+
+void H3WtCapsuleCallback::onMaxStreamsUni(WTMaxStreamsCapsule c) noexcept {
+  XLOG(DBG6) << __func__ << "; maxStreams=" << c.maximumStreams;
+  h3Wt_.onMaxStreams(WtStreamManager::MaxStreamsUni{c.maximumStreams});
+}
+
+void H3WtCapsuleCallback::onDrainSession(
+    DrainWebTransportSessionCapsule) noexcept {
+  XLOG(DBG6) << __func__;
+  h3Wt_.onDrainSession({});
+}
+
+void H3WtCapsuleCallback::onCloseSession(
+    CloseWebTransportSessionCapsule c) noexcept {
+  XLOG(DBG6) << __func__ << "; err=" << c.applicationErrorCode
+             << "; msg=" << c.applicationErrorMessage;
+  h3Wt_.onCloseSession(
+      {.err = c.applicationErrorCode, .msg = c.applicationErrorMessage});
+}
+
+void H3WtCapsuleCallback::onDatagram(DatagramCapsule c) noexcept {
+  XLOG(DBG6) << __func__;
+  h3Wt_.onDatagram(std::move(c.httpDatagramPayload));
+}
+
+void H3WtCapsuleCallback::onConnectionError(
+    CapsuleCodec::ErrorCode error) noexcept {
+  XLOG(DBG4) << __func__ << "; err=" << uint64_t(error);
+  onCloseSession({.applicationErrorCode = uint32_t(error),
+                  .applicationErrorMessage = "capsule parse error"});
+}
+
 } // namespace proxygen

proxygen/lib/http/webtransport/QuicWtSession.h

@@ -296,4 +296,54 @@ class H3WtSession final : public QuicWtSessionBase {
   uint64_t connectStreamId_;
 };
 
+/**
+ * H3WtCapsuleCallback receives parsed capsule events from the
+ * WebTransportCapsuleCodec on the http/3 CONNECT stream and forwards the
+ * relevant subset (MaxData, MaxStreams(Uni|Bidi), DrainSession, CloseSession,
+ * Datagram) to the owning H3WtSession.
+ *
+ * A datagram may reach an http/3 endpoint either as a quic datagram (RFC9297)
+ * or as a DATAGRAM capsule on the CONNECT stream, so both paths land on
+ * H3WtSession::onDatagram.
+ */
+struct H3WtCapsuleCallback : WebTransportCapsuleCodec::Callback {
+  explicit H3WtCapsuleCallback(H3WtSession& session) noexcept;
+
+  void onMaxData(WTMaxDataCapsule c) noexcept override;
+  void onMaxStreamsBidi(WTMaxStreamsCapsule c) noexcept override;
+  void onMaxStreamsUni(WTMaxStreamsCapsule c) noexcept override;
+  void onDrainSession(DrainWebTransportSessionCapsule) noexcept override;
+  void onCloseSession(CloseWebTransportSessionCapsule c) noexcept override;
+  void onDatagram(DatagramCapsule c) noexcept override;
+  void onConnectionError(CapsuleCodec::ErrorCode error) noexcept override;
+
+  /**
+   * Streams and per-stream flow control are carried by real quic streams over
+   * http/3, so the capsules describing them are only meaningful over http/2.
+   */
+  void onPadding(PaddingCapsule) noexcept override {
+  }
+  void onResetStream(WTResetStreamCapsule) noexcept override {
+  }
+  void onStopSending(WTStopSendingCapsule) noexcept override {
+  }
+  void onStream(WTStreamCapsule) noexcept override {
+  }
+  void onMaxStreamData(WTMaxStreamDataCapsule) noexcept override {
+  }
+  void onDataBlocked(WTDataBlockedCapsule) noexcept override {
+  }
+  void onStreamDataBlocked(WTStreamDataBlockedCapsule) noexcept override {
+  }
+  void onStreamsBlockedBidi(WTStreamsBlockedCapsule) noexcept override {
+  }
+  void onStreamsBlockedUni(WTStreamsBlockedCapsule) noexcept override {
+  }
+  void onCapsule(uint64_t, uint64_t) noexcept override {
+  }
+
+ private:
+  H3WtSession& h3Wt_;
+};
+
 } // namespace proxygen

proxygen/lib/http/webtransport/test/QuicWtSessionTest.cpp

@@ -8,6 +8,7 @@
 
 #include <folly/portability/GMock.h>
 #include <folly/portability/GTest.h>
+#include <proxygen/lib/http/codec/webtransport/WebTransportFramer.h>
 #include <proxygen/lib/http/session/test/MockQuicSocketDriver.h>
 #include <proxygen/lib/http/webtransport/QuicWtSession.h>
 #include <proxygen/lib/http/webtransport/test/Mocks.h>
@@ -921,6 +922,33 @@ TEST_F(H3WtSessionTest, CreateUniBidiStream) {
   }
 }
 
+/**
+ * A malformed capsule on the CONNECT stream is unrecoverable -- we can no
+ * longer trust the framing -- so H3WtCapsuleCallback tears the wt session down
+ * rather than ignoring it.
+ */
+TEST_F(H3WtSessionTest, CapsuleParseErrorClosesSession) {
+  expectedWtHandlerErr_ = uint32_t(CapsuleCodec::ErrorCode::PARSE_UNDERFLOW);
+
+  H3WtCapsuleCallback capsuleCb{*session_};
+  WebTransportCapsuleCodec codec{&capsuleCb, CodecVersion::H3};
+
+  // a well formed MAX_DATA capsule whose length byte is rewritten to announce
+  // an 8-byte varint that is not there, so the codec reports a parse error
+  folly::IOBufQueue queue{folly::IOBufQueue::cacheChainLength()};
+  writeWTMaxData(queue, WTMaxDataCapsule{100});
+  auto buf = queue.move();
+  buf->writableData()[5] = 0xFF;
+
+  codec.onIngress(std::move(buf), true);
+
+  // the handler is notified via the fixture's onSessionEnd expectation, and the
+  // session queues a CLOSE_SESSION for the connect stream
+  EXPECT_FALSE(connectStreamCb_.events.empty());
+  EXPECT_TRUE(std::holds_alternative<detail::WtStreamManager::CloseSession>(
+      connectStreamCb_.events.back()));
+}
+
 TEST_F(H3WtSessionTest, AcquireIngressStream) {
   // client-initiated stream ids (server is the local endpoint)
   constexpr uint64_t kClientBidiId = 0;