From 479eb5574195764e80e6cedebef669f6baa85083 Mon Sep 17 00:00:00 2001 From: "meta-codesync[bot]" <215208954+meta-codesync[bot]@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:51:55 -0700 Subject: [PATCH] Re-sync with internal repository (#645) The internal and external repositories are out of sync. This Pull Request attempts to brings them back in sync by patching the GitHub repository. Please carefully review this patch. You must disable ShipIt for your project in order to merge this pull request. DO NOT IMPORT this pull request. Instead, merge it directly on GitHub using the MERGE BUTTON. Re-enable ShipIt after merging. fbshipit-source-id: e24e32453662b79b80308343ceefcb628225c94c Co-authored-by: Facebook GitHub Bot --- proxygen/lib/http/codec/HTTPBinaryCodec.cpp | 78 ++-- proxygen/lib/http/codec/HTTPBinaryCodec.h | 3 + .../http/codec/test/HTTPBinaryCodecTest.cpp | 363 +++++++++++++++++- proxygen/lib/http/session/HQSession.cpp | 34 +- proxygen/lib/http/session/HTTPTransaction.cpp | 19 +- .../session/test/HQDownstreamSessionTest.cpp | 93 +++++ .../test/HTTPTransactionWebTransportTest.cpp | 111 +++++- .../http/webtransport/WebTransportImpl.cpp | 41 +- .../lib/http/webtransport/WebTransportImpl.h | 19 + 9 files changed, 707 insertions(+), 54 deletions(-) diff --git a/proxygen/lib/http/codec/HTTPBinaryCodec.cpp b/proxygen/lib/http/codec/HTTPBinaryCodec.cpp index a9d031b59..1fc9db618 100644 --- a/proxygen/lib/http/codec/HTTPBinaryCodec.cpp +++ b/proxygen/lib/http/codec/HTTPBinaryCodec.cpp @@ -19,6 +19,7 @@ break; \ } else if ((parseResult).parseResultState_ == \ ParseResultState::WAITING_FOR_MORE_DATA) { \ + parsed += (parseResult).bytesParsed_; \ parserWaitingForMoreData_ = true; \ break; \ } @@ -221,17 +222,20 @@ ParseResult HTTPBinaryCodec::parseSingleHeaderHelper( size_t& remaining, uint64_t maxLength, size_t& numHeaders) { - std::string headerName; - auto headerNameRes = parseKnownLengthString( - cursor, remaining, maxLength, "headerName", headerName); - if (headerNameRes.parseResultState_ == ParseResultState::ERROR || - headerNameRes.parseResultState_ == - ParseResultState::WAITING_FOR_MORE_DATA) { - return headerNameRes; + if (!pendingFieldName_) { + std::string headerName; + auto headerNameRes = parseKnownLengthString( + cursor, remaining, maxLength, "headerName", headerName); + if (headerNameRes.parseResultState_ == ParseResultState::ERROR || + headerNameRes.parseResultState_ == + ParseResultState::WAITING_FOR_MORE_DATA) { + return headerNameRes; + } + parsed += headerNameRes.bytesParsed_; + remaining -= headerNameRes.bytesParsed_; + maxLength -= std::min(maxLength, headerNameRes.bytesParsed_); + pendingFieldName_.emplace(std::move(headerName)); } - parsed += headerNameRes.bytesParsed_; - remaining -= headerNameRes.bytesParsed_; - maxLength -= std::min(maxLength, headerNameRes.bytesParsed_); std::string headerValue; auto headerValueRes = parseKnownLengthString( @@ -244,6 +248,9 @@ ParseResult HTTPBinaryCodec::parseSingleHeaderHelper( parsed += headerValueRes.bytesParsed_; remaining -= headerValueRes.bytesParsed_; + CHECK(pendingFieldName_); + auto headerName = std::move(*pendingFieldName_); + pendingFieldName_.reset(); if (!decodeInfo.onHeader(proxygen::HPACKHeaderName(headerName), headerValue) || !decodeInfo.parsingError.empty()) { @@ -287,20 +294,31 @@ ParseResult HTTPBinaryCodec::parseKnownLengthHeadersHelper( return ParseResult(ParseResultState::WAITING_FOR_MORE_DATA); } + const auto fieldSectionLength = static_cast(lengthOfHeaders->first); + folly::io::Cursor fieldSectionCursor(cursor, fieldSectionLength); + size_t fieldSectionParsed = 0; + size_t fieldSectionRemaining = fieldSectionLength; size_t numHeaders = 0; - while (parsed < lengthOfHeaders->first) { - auto result = parseSingleHeaderHelper(cursor, + while (fieldSectionParsed < fieldSectionLength) { + auto result = parseSingleHeaderHelper(fieldSectionCursor, decodeInfo, - parsed, - remaining, - maxFieldSectionSize_, + fieldSectionParsed, + fieldSectionRemaining, + fieldSectionRemaining, numHeaders); - if (result.parseResultState_ == ParseResultState::ERROR || - result.parseResultState_ == ParseResultState::WAITING_FOR_MORE_DATA) { + if (result.parseResultState_ == ParseResultState::ERROR) { return result; } + if (result.parseResultState_ == ParseResultState::WAITING_FOR_MORE_DATA) { + return ParseResult( + fmt::format("{} field exceeds declared section length of {}", + isTrailers ? "Trailer" : "Header", + lengthOfHeaders->first)); + } } + cursor.skip(fieldSectionLength); + parsed += fieldSectionParsed; return ParseResult(parsed); } @@ -314,21 +332,28 @@ ParseResult HTTPBinaryCodec::parseIndeterminateLengthHeadersHelper( auto currentByte = cursor.peek().data(); size_t numHeaders = 0; // Continue parsing headers until we reach the Content Terminator field (0) - while (currentByte != nullptr && *currentByte != 0x00) { + while (currentByte != nullptr && + (pendingFieldName_ || *currentByte != 0x00)) { // A field whose declared length cannot fit in what is left of the section // has to be rejected on sight: waiting for it to arrive would retain the // fields already parsed plus the whole declaration const uint64_t sectionBudget = - maxFieldSectionSize_ - std::min(parsed, maxFieldSectionSize_); + maxFieldSectionSize_ - + std::min(indeterminateFieldSectionBytesParsed_ + parsed, + maxFieldSectionSize_); auto result = parseSingleHeaderHelper( cursor, decodeInfo, parsed, remaining, sectionBudget, numHeaders); - if (result.parseResultState_ == ParseResultState::ERROR || - result.parseResultState_ == ParseResultState::WAITING_FOR_MORE_DATA) { + if (result.parseResultState_ == ParseResultState::ERROR) { return result; } + if (result.parseResultState_ == ParseResultState::WAITING_FOR_MORE_DATA) { + indeterminateFieldSectionBytesParsed_ += parsed; + return ParseResult(parsed, ParseResultState::WAITING_FOR_MORE_DATA); + } // There is no declared length to check against, so bound the section by // what has actually been seen - if (parsed > maxFieldSectionSize_) { + if (indeterminateFieldSectionBytesParsed_ > maxFieldSectionSize_ || + parsed > maxFieldSectionSize_ - indeterminateFieldSectionBytesParsed_) { return ParseResult( fmt::format("Unterminated {} section exceeds the maximum field " "section size of {}", @@ -339,7 +364,8 @@ ParseResult HTTPBinaryCodec::parseIndeterminateLengthHeadersHelper( // waiting for more data since we haven't seen a Content Terminator field // yet if (cursor.isAtEnd()) { - return ParseResult(ParseResultState::WAITING_FOR_MORE_DATA); + indeterminateFieldSectionBytesParsed_ += parsed; + return ParseResult(parsed, ParseResultState::WAITING_FOR_MORE_DATA); } currentByte = cursor.peek().data(); } @@ -347,6 +373,7 @@ ParseResult HTTPBinaryCodec::parseIndeterminateLengthHeadersHelper( parsed++; remaining--; cursor.skip(1); + indeterminateFieldSectionBytesParsed_ = 0; return ParseResult(parsed); } @@ -613,7 +640,8 @@ size_t HTTPBinaryCodec::onIngress(const folly::IOBuf& buf) { void HTTPBinaryCodec::onIngressEOF() { if (!parseError_ && - (!bufferedIngress_.empty() || remainingContentLength_.value_or(0) > 0)) { + (!bufferedIngress_.empty() || remainingContentLength_.value_or(0) > 0 || + indeterminateFieldSectionBytesParsed_ > 0)) { // Case where the ingress EOF is received before the entire message is // parsed callback_->onError(ingressTxnID_, @@ -632,7 +660,7 @@ void HTTPBinaryCodec::onIngressEOF() { "contains framing indicator")); return; } - if (state_ == ParseState::HEADERS_SECTION) { + if (!parseError_ && state_ == ParseState::HEADERS_SECTION) { // Case where the sent message only contains control data and no headers // nor body callback_->onHeadersComplete(ingressTxnID_, std::move(decodeInfo_.msg)); diff --git a/proxygen/lib/http/codec/HTTPBinaryCodec.h b/proxygen/lib/http/codec/HTTPBinaryCodec.h index c1f061504..ea5106216 100644 --- a/proxygen/lib/http/codec/HTTPBinaryCodec.h +++ b/proxygen/lib/http/codec/HTTPBinaryCodec.h @@ -236,6 +236,9 @@ class HTTPBinaryCodec : public HTTPCodec { bool parserWaitingForMoreData_{false}; bool parserPaused_; folly::Optional parseError_{folly::none}; + // Complete field bytes trimmed from the current indeterminate section. + size_t indeterminateFieldSectionBytesParsed_{0}; + folly::Optional pendingFieldName_{folly::none}; enum class FramingIndicator : uint8_t { REQUEST_KNOWN_LENGTH = 0, diff --git a/proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp b/proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp index ffb157ed4..3ba9dce6a 100644 --- a/proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp +++ b/proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp @@ -58,9 +58,17 @@ class HTTPBinaryCodecForTest : public HTTPBinaryCodec { return *msgBody_; } + size_t getPendingFieldCount() const { + return decodeInfo_.msg ? decodeInfo_.msg->getHeaders().size() : 0; + } + size_t getBufferedIngressSize() const { return bufferedIngress_.chainLength(); } + + const std::string* getPendingFieldName() const { + return pendingFieldName_ ? &*pendingFieldName_ : nullptr; + } }; namespace { @@ -89,6 +97,19 @@ void writeRequestPreamble(folly::io::QueueAppender& appender, writeVarintString(appender, "/"); } +void writeOverrunningKnownLengthFieldSection( + folly::io::QueueAppender& appender) { + folly::IOBufQueue fieldSection{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender fieldAppender(&fieldSection, 64); + writeVarintString(fieldAppender, "x-first"); + writeVarintString(fieldAppender, "first-value"); + writeVarintString(fieldAppender, "x-last"); + writeVarint(fieldAppender, 64); + + writeVarint(appender, fieldSection.chainLength()); + appender.insert(fieldSection.move()); +} + std::unique_ptr makeFiller(size_t size) { auto buf = folly::IOBuf::create(size); buf->append(size); @@ -359,11 +380,12 @@ TEST_F(HttpBinaryUpstreamCodecTest, testParseKnownLengthHeadersEmpty) { .bytesParsed_, 1); } -TEST_F(HttpBinaryUpstreamCodecTest, testParseKnownLengthHeadersWaiting) { +TEST_F(HttpBinaryUpstreamCodecTest, + testParseKnownLengthHeadersOverrunsDeclaredSection) { // Format is `**user-agent*curl/7.16.3 libcurl/7.16.3 OpenSSL/0.9.7l // zlib/1.2.3*host*www.example.com*accept-language.en, mi` where the `*` after - // accept-language is value 7 instead of 6 which should cause parsing to wait - // to get more data + // accept-language is value 7 instead of 6, which exceeds the declared + // section boundary. const std::vector invalidHeadersLength{ 0x40, 0x6c, 0x0a, 0x75, 0x73, 0x65, 0x72, 0x2d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x34, 0x63, 0x75, 0x72, 0x6c, 0x2f, 0x37, 0x2e, 0x31, @@ -391,7 +413,7 @@ TEST_F(HttpBinaryUpstreamCodecTest, testParseKnownLengthHeadersWaiting) { decodeInfo, /*knownLength=*/true) .parseResultState_, - ParseResultState::WAITING_FOR_MORE_DATA); + ParseResultState::ERROR); // Format is `**a*b` where the first `*` represents a too long length const std::vector invalidHeadersUnderflow{ @@ -796,7 +818,9 @@ TEST_F(HttpBinaryDownstreamCodecTest, testOnIngressFailureMalformedMessage) { // Check onError was called with the correct error EXPECT_EQ(std::string(callback.lastParseError.get()->what()), - "Incomplete message received"); + "Invalid Message: Header field exceeds declared section length of " + "108"); + EXPECT_EQ(callback.headersComplete, 0); } TEST_F(HttpBinaryDownstreamCodecTest, testOnIngressFailureIncompleteMessage) { @@ -1240,6 +1264,245 @@ TEST_F(HttpBinaryDownstreamCodecTest, testPartialChunkIsNotOverCounted) { EXPECT_EQ(callback.data_.move()->to(), first + second); } +TEST_F(HttpBinaryDownstreamCodecTest, testSplitHeaderFieldIsDecodedOnce) { + const std::string splitValue = "split-value"; + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 256); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, "x-first"); + writeVarintString(appender, "first-value"); + writeVarintString(appender, "x-split"); + writeVarint(appender, splitValue.size()); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + + for (size_t i = 0; i < splitValue.size(); i++) { + auto chunk = folly::IOBuf::copyBuffer(splitValue.data() + i, 1); + binaryCodecIndeterminateLength_->onIngress(*chunk); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), + i + 1 == splitValue.size() ? 2 : 1); + } + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + folly::IOBufQueue terminators{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender terminatorAppender(&terminators, 4); + writeVarint(terminatorAppender, 0 /* field section terminator */); + writeVarint(terminatorAppender, 0 /* empty content field */); + writeVarint(terminatorAppender, 0 /* content terminator */); + writeVarint(terminatorAppender, 0 /* trailer section terminator */); + binaryCodecIndeterminateLength_->onIngress(*terminators.front()); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().size(), 2); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty("x-first"), + "first-value"); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty("x-split"), splitValue); +} + +TEST_F(HttpBinaryDownstreamCodecTest, testSplitTrailerFieldIsDecodedOnce) { + const std::string splitValue = "split-value"; + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 256); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarint(appender, 0 /* field section terminator */); + writeVarint(appender, 0 /* empty content field */); + writeVarint(appender, 0 /* content terminator */); + writeVarintString(appender, "x-first"); + writeVarintString(appender, "first-value"); + writeVarintString(appender, "x-split"); + writeVarint(appender, splitValue.size()); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + + for (size_t i = 0; i < splitValue.size(); i++) { + auto chunk = folly::IOBuf::copyBuffer(splitValue.data() + i, 1); + binaryCodecIndeterminateLength_->onIngress(*chunk); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), + i + 1 == splitValue.size() ? 2 : 1); + } + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + auto terminator = folly::IOBuf::copyBuffer("\0", 1); + binaryCodecIndeterminateLength_->onIngress(*terminator); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + ASSERT_NE(callback.msg->getTrailers(), nullptr); + EXPECT_EQ(callback.msg->getTrailers()->size(), 2); + EXPECT_EQ(callback.msg->getTrailers()->getSingleOrEmpty("x-first"), + "first-value"); + EXPECT_EQ(callback.msg->getTrailers()->getSingleOrEmpty("x-split"), + splitValue); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testSplitHeaderLengthVarintIsDecodedOnce) { + const std::string splitValue(64, 'a'); + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 256); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, "x-first"); + writeVarintString(appender, "first-value"); + writeVarintString(appender, "x-split"); + writeVarint(appender, splitValue.size()); + + auto head = message.split(message.chainLength() - 1); + auto lengthSuffix = message.move(); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*head); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + + binaryCodecIndeterminateLength_->onIngress(*lengthSuffix); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + + auto value = folly::IOBuf::copyBuffer(splitValue.data(), splitValue.size()); + binaryCodecIndeterminateLength_->onIngress(*value); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 2); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + auto terminators = folly::IOBuf::copyBuffer("\0\0\0\0", 4); + binaryCodecIndeterminateLength_->onIngress(*terminators); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty("x-split"), splitValue); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testBytePacedLargeFieldNameIsTrimmedOnce) { + constexpr size_t kOhttpFieldSectionSize = size_t{64} * 1024; + constexpr size_t kFieldPartSize = kOhttpFieldSectionSize / 2 - 16; + const std::string headerName(kFieldPartSize, 'x'); + const std::string headerValue(kFieldPartSize, 'v'); + + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 256); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarint(appender, headerName.size()); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setMaxFieldSectionSize( + kOhttpFieldSectionSize); + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + const auto sendByte = [&](const std::string& input, size_t offset) { + auto chunk = folly::IOBuf::copyBuffer(input.data() + offset, 1); + binaryCodecIndeterminateLength_->onIngress(*chunk); + }; + for (size_t i = 0; i < headerName.size(); i++) { + sendByte(headerName, i); + } + + const auto* pendingFieldName = + binaryCodecIndeterminateLength_->getPendingFieldName(); + ASSERT_NE(pendingFieldName, nullptr); + EXPECT_EQ(*pendingFieldName, headerName); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 0); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + folly::IOBufQueue valueLength{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender valueLengthAppender(&valueLength, 8); + writeVarint(valueLengthAppender, headerValue.size()); + const auto valueLengthSize = valueLength.chainLength(); + binaryCodecIndeterminateLength_->onIngress(*valueLength.front()); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), + valueLengthSize); + + for (size_t i = 0; i + 1 < headerValue.size(); i++) { + sendByte(headerValue, i); + } + pendingFieldName = binaryCodecIndeterminateLength_->getPendingFieldName(); + ASSERT_NE(pendingFieldName, nullptr); + EXPECT_EQ(*pendingFieldName, headerName); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), + valueLengthSize + headerValue.size() - 1); + sendByte(headerValue, headerValue.size() - 1); + + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldName(), nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + auto terminators = folly::IOBuf::copyBuffer("\0\0\0\0", 4); + binaryCodecIndeterminateLength_->onIngress(*terminators); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty(headerName), + headerValue); +} + +TEST_F(HttpBinaryDownstreamCodecTest, testCachedFieldNameAllowsEmptyValue) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, "x-empty"); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + ASSERT_NE(binaryCodecIndeterminateLength_->getPendingFieldName(), nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + auto emptyValueAndTerminators = folly::IOBuf::copyBuffer("\0\0\0\0\0", 5); + binaryCodecIndeterminateLength_->onIngress(*emptyValueAndTerminators); + + EXPECT_EQ(callback.lastParseError, nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldName(), nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().size(), 1); + EXPECT_TRUE(callback.msg->getHeaders().exists("x-empty")); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty("x-empty"), ""); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testKnownLengthHeaderFieldCannotExceedDeclaredSection) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 0 /* request, known length */); + writeOverrunningKnownLengthFieldSection(appender); + + FakeHTTPCodecCallback callback; + binaryCodecKnownLength_->setCallback(&callback); + binaryCodecKnownLength_->onIngress(*message.front()); + + ASSERT_NE(callback.lastParseError, nullptr); + EXPECT_EQ(std::string(callback.lastParseError->what()), + "Invalid Message: Failure to parse: headerValue of declared length " + "64 exceeds the remaining field section budget of 2"); + EXPECT_EQ(binaryCodecKnownLength_->getPendingFieldCount(), 1); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testKnownLengthTrailerFieldCannotExceedDeclaredSection) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 0 /* request, known length */); + writeVarint(appender, 0 /* empty header section */); + writeVarint(appender, 0 /* empty content */); + writeOverrunningKnownLengthFieldSection(appender); + + FakeHTTPCodecCallback callback; + binaryCodecKnownLength_->setCallback(&callback); + binaryCodecKnownLength_->onIngress(*message.front()); + + ASSERT_NE(callback.lastParseError, nullptr); + EXPECT_EQ(std::string(callback.lastParseError->what()), + "Invalid Message: Failure to parse: headerValue of declared length " + "64 exceeds the remaining field section budget of 2"); + EXPECT_EQ(binaryCodecKnownLength_->getPendingFieldCount(), 1); +} + // Field section lengths are QUIC varints, so a peer can declare up to 2^62-1 // bytes. The declaration must be rejected up front instead of buffering // ingress until the section is complete. @@ -1306,15 +1569,54 @@ TEST_F(HttpBinaryDownstreamCodecTest, writeVarintString(fieldsAppender, "x-pad"); writeVarintString(fieldsAppender, padValue); } + constexpr size_t kSuccessfulBatchCount = 2; + binaryCodecIndeterminateLength_->setMaxFieldSectionSize( + fields.chainLength() * kSuccessfulBatchCount); auto fieldsBuf = fields.move(); - for (size_t i = 0; i < kFillerChunkCount && !callback.lastParseError; i++) { + for (size_t i = 0; i < kSuccessfulBatchCount; i++) { binaryCodecIndeterminateLength_->onIngress(*fieldsBuf); + ASSERT_EQ(callback.lastParseError, nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); } + binaryCodecIndeterminateLength_->onIngress(*fieldsBuf); EXPECT_NE(callback.lastParseError, nullptr); } +TEST_F(HttpBinaryDownstreamCodecTest, + testIndeterminateFieldSectionLimitResetsForTrailers) { + const std::string fieldName = "x-near-cap"; + const std::string fieldValue(128, 'a'); + const auto fieldSize = + quic::getQuicIntegerSize(fieldName.size()).value() + fieldName.size() + + quic::getQuicIntegerSize(fieldValue.size()).value() + fieldValue.size(); + + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 512); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, fieldName); + writeVarintString(appender, fieldValue); + writeVarint(appender, 0 /* field section terminator */); + writeVarint(appender, 0 /* empty content field */); + writeVarint(appender, 0 /* content terminator */); + writeVarintString(appender, fieldName); + writeVarintString(appender, fieldValue); + writeVarint(appender, 0 /* trailer section terminator */); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setMaxFieldSectionSize(fieldSize); + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty(fieldName), fieldValue); + ASSERT_NE(callback.msg->getTrailers(), nullptr); + EXPECT_EQ(callback.msg->getTrailers()->getSingleOrEmpty(fieldName), + fieldValue); +} + TEST_F(HttpBinaryDownstreamCodecTest, testMaxFieldSectionSizeIsConfigurable) { const std::string headerValue(256, 'a'); folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; @@ -1432,4 +1734,53 @@ TEST_F(HttpBinaryDownstreamCodecTest, testTruncatedContentIsIncompleteAtEOF) { "Incomplete message received"); } +TEST_F(HttpBinaryDownstreamCodecTest, + testTrimmedFieldWithoutTerminatorIsIncompleteAtEOF) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, "x-complete"); + writeVarintString(appender, "value"); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + EXPECT_EQ(callback.lastParseError, nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + binaryCodecIndeterminateLength_->onIngressEOF(); + + ASSERT_NE(callback.lastParseError, nullptr); + EXPECT_EQ(std::string(callback.lastParseError->what()), + "Incomplete message received"); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testTrimmedTrailerWithoutTerminatorIsIncompleteAtEOF) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarint(appender, 0 /* field section terminator */); + writeVarint(appender, 0 /* empty content field */); + writeVarint(appender, 0 /* content terminator */); + writeVarintString(appender, "x-complete"); + writeVarintString(appender, "value"); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + EXPECT_EQ(callback.lastParseError, nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + binaryCodecIndeterminateLength_->onIngressEOF(); + + ASSERT_NE(callback.lastParseError, nullptr); + EXPECT_EQ(std::string(callback.lastParseError->what()), + "Incomplete message received"); +} + } // namespace proxygen::test diff --git a/proxygen/lib/http/session/HQSession.cpp b/proxygen/lib/http/session/HQSession.cpp index a067f40a6..18fcca211 100644 --- a/proxygen/lib/http/session/HQSession.cpp +++ b/proxygen/lib/http/session/HQSession.cpp @@ -3594,12 +3594,19 @@ void HQSession::deliverWTStream(HQStreamTransportBase* connectStream, return; } connectStream->txn_.getWebTransport(); // lazy-init webTransportImpl_ + // The handler runs below and can end the txn, freeing the handle. + HTTPTransaction::DestructorGuard dg(&connectStream->txn_); + WebTransportImpl::StreamReadHandle* readHandle = nullptr; if (sock_->isBidirectionalStream(streamID)) { - auto handle = connectStream->txn_.onWebTransportBidiStream(streamID); - sock_->setReadCallback(streamID, handle.readHandle); + readHandle = + connectStream->txn_.onWebTransportBidiStream(streamID).readHandle; } else { - auto* handle = connectStream->txn_.onWebTransportUniStream(streamID); - sock_->setReadCallback(streamID, handle); + readHandle = connectStream->txn_.onWebTransportUniStream(streamID); + } + // No handle means the handler ended the session; an unregistered one means + // it called stopSending. + if (readHandle && readHandle->readCallbackRegistered()) { + sock_->setReadCallback(streamID, readHandle); } } @@ -3676,7 +3683,16 @@ void HQSession::deliverPendingWTStreams(quic::StreamId sessionID) noexcept { VLOG(4) << "delivering " << entry->streams.size() << " buffered WT streams wt-sess-id=" << sessionID << " sess=" << *this; + // The handler may end the CONNECT partway through the loop. The guard keeps + // connectStream alive so the streams behind it can be rejected. + HTTPTransaction::DestructorGuard dg(&connectStream->txn_); for (auto id : entry->streams) { + if (connectStream->txn_.isIngressComplete()) { + VLOG(4) << "WT connect stream ended mid-drain, rejecting id=" << id + << " wt-sess-id=" << sessionID << " sess=" << *this; + rejectStream(id, kBufferedRejectedErr); + continue; + } deliverWTStream(connectStream, id); } } @@ -4037,7 +4053,15 @@ HQSession::HQStreamTransport::stopReadingWebTransportIngress( } auto res = session_.sock_->setReadCallback(id, nullptr, quicErrorCode); if (res.hasError()) { - return folly::makeUnexpected(WebTransport::ErrorCode::GENERIC_ERROR); + // setReadCallback only sends STOP_SENDING if a read callback is + // installed, which it isn't until deliverWTStream runs. + if (!quicErrorCode) { + return folly::makeUnexpected(WebTransport::ErrorCode::GENERIC_ERROR); + } + auto ssRes = session_.sock_->stopSending(id, *quicErrorCode); + if (ssRes.hasError()) { + return folly::makeUnexpected(WebTransport::ErrorCode::GENERIC_ERROR); + } } } return folly::unit; diff --git a/proxygen/lib/http/session/HTTPTransaction.cpp b/proxygen/lib/http/session/HTTPTransaction.cpp index d20e9fcb5..542310d7b 100644 --- a/proxygen/lib/http/session/HTTPTransaction.cpp +++ b/proxygen/lib/http/session/HTTPTransaction.cpp @@ -1825,12 +1825,19 @@ WebTransportImpl::BidiStreamHandle HTTPTransaction::onWebTransportBidiStream( id, WebTransport::kInternalError); return {.readHandle = nullptr, .writeHandle = nullptr}; } + // The handler may close the session or abort the txn, freeing the handles + // and possibly this. + DestructorGuard dg(this); refreshTimeout(); auto handle = webTransportImpl_->onWebTransportBidiStream(id); handler_->onWebTransportBidiStream( id, {.readHandle = handle.readHandle, .writeHandle = handle.writeHandle}); - // what if the handler killed handle (stop sending/fin/rst)? - return handle; + if (!handler_ || isIngressComplete()) { + // The handler ended the txn; dg's release frees the handles with it. + return {.readHandle = nullptr, .writeHandle = nullptr}; + } + return {.readHandle = webTransportImpl_->getReadHandle(id), + .writeHandle = webTransportImpl_->getWriteHandle(id)}; } WebTransportImpl::StreamReadHandle* HTTPTransaction::onWebTransportUniStream( @@ -1840,12 +1847,16 @@ WebTransportImpl::StreamReadHandle* HTTPTransaction::onWebTransportUniStream( id, WebTransport::kInternalError); return nullptr; } + // See onWebTransportBidiStream. + DestructorGuard dg(this); refreshTimeout(); auto handle = webTransportImpl_->onWebTransportUniStream(id); handler_->onWebTransportUniStream(id, handle); - // what if the handler killed handle (stop sending)? - return handle; + if (!handler_ || isIngressComplete()) { + return nullptr; + } + return webTransportImpl_->getReadHandle(id); } bool HTTPTransaction::onWebTransportStopSending(HTTPCodec::StreamID id, diff --git a/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp b/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp index 995f17e1e..8e1646f22 100644 --- a/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp +++ b/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp @@ -3940,6 +3940,99 @@ TEST_P(HQDownstreamSessionTestWebTransport, BufferedWTStreamRejectedOnClose) { EXPECT_EQ(*socketDriver_->streams_[wtStreamId].error, bufferedRejectedErr); } +// The handler aborts the CONNECT on the first buffered stream, which would +// destroy the stream the drain loop is walking. The rest get rejected. +TEST_P(HQDownstreamSessionTestWebTransport, + BufferedWTStreamConnectAbortedDuringDrain) { + const quic::StreamId sessionId = 0; + const quic::StreamId wtStreamId1 = 14; + const quic::StreamId wtStreamId2 = 18; + + sendSettings(); + queuePeerWTUniStream(socketDriver_.get(), wtStreamId1, sessionId); + queuePeerWTUniStream(socketDriver_.get(), wtStreamId2, sessionId); + flushRequestsAndLoopN(1); + EXPECT_FALSE(socketDriver_->streams_[wtStreamId1].error.has_value()); + EXPECT_FALSE(socketDriver_->streams_[wtStreamId2].error.has_value()); + + HTTPMessage req; + req.setHTTPVersion(1, 1); + req.setUpgradeProtocol("webtransport"); + req.setMethod(HTTPMethod::CONNECT); + req.setURL("/webtransport"); + req.getHeaders().set(HTTP_HEADER_HOST, "www.facebook.com"); + auto actualSessionId = sendRequest(req, /*eom=*/false); + ASSERT_EQ(actualSessionId, sessionId); + + auto handler = addSimpleStrictHandler(); + handler->expectHeaders([&]() { + HTTPMessage resp; + resp.setStatusCode(200); + handler->txn_->sendHeaders(resp); + EXPECT_NE(handler->txn_->getWebTransport(), nullptr); + }); + // Only the first buffered stream reaches the handler. + EXPECT_CALL(*handler, onWebTransportUniStream(_, _)) + .Times(1) + .WillOnce([&](HTTPCodec::StreamID, WebTransport::StreamReadHandle*) { + handler->txn_->sendAbort(); + }); + handler->expectDetachTransaction(); + flushRequestsAndLoopN(3); + + const uint64_t bufferedRejectedErr = + WebTransport::toHTTPErrorCode(WebTransport::kBufferedStreamRejected); + ASSERT_TRUE(socketDriver_->streams_[wtStreamId2].error.has_value()); + EXPECT_EQ(*socketDriver_->streams_[wtStreamId2].error, bufferedRejectedErr); + + hqSession_->dropConnection(); + flushRequestsAndLoop(); +} + +// The handler stops reading from inside onWebTransportUniStream, so +// deliverWTStream must not turn around and register the handle with quic. +TEST_P(HQDownstreamSessionTestWebTransport, StopSendingFromOnNewUniStream) { + const quic::StreamId sessionId = 0; + const quic::StreamId wtStreamId = 14; + + sendSettings(); + HTTPMessage req; + req.setHTTPVersion(1, 1); + req.setUpgradeProtocol("webtransport"); + req.setMethod(HTTPMethod::CONNECT); + req.setURL("/webtransport"); + req.getHeaders().set(HTTP_HEADER_HOST, "www.facebook.com"); + auto actualSessionId = sendRequest(req, /*eom=*/false); + ASSERT_EQ(actualSessionId, sessionId); + + auto handler = addSimpleStrictHandler(); + handler->expectHeaders([&]() { + HTTPMessage resp; + resp.setStatusCode(200); + handler->txn_->sendHeaders(resp); + EXPECT_NE(handler->txn_->getWebTransport(), nullptr); + }); + flushRequestsAndLoopN(3); + + EXPECT_CALL(*handler, onWebTransportUniStream(wtStreamId, _)) + .WillOnce( + [](HTTPCodec::StreamID, WebTransport::StreamReadHandle* readHandle) { + readHandle->stopSending(WebTransport::kInternalError); + }); + queuePeerWTUniStream(socketDriver_.get(), wtStreamId, sessionId); + flushRequestsAndLoopN(2); + + // The handle dies with the session, so it must not be left registered. The + // STOP_SENDING still has to go out, via QuicSocket::stopSending(). + EXPECT_EQ(socketDriver_->streams_[wtStreamId].readCB, nullptr); + EXPECT_TRUE(socketDriver_->streams_[wtStreamId].error.has_value()); + + handler->expectDetachTransaction(); + handler->txn_->sendAbort(); + hqSession_->dropConnection(); + flushRequestsAndLoop(); +} + INSTANTIATE_TEST_SUITE_P(HQDownstreamSessionTest, HQDownstreamSessionTestWebTransport, Values([] { diff --git a/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp b/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp index b9125b615..55c51c908 100644 --- a/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp +++ b/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp @@ -512,7 +512,8 @@ TEST_F(HTTPTransactionWebTransportTest, BidiStreamEdgeCases) { // it gets stopReadingWebTransportIngress when the EOF is read out EXPECT_CALL(transport_, - stopReadingWebTransportIngress(0, folly::Optional())); + stopReadingWebTransportIngress(0, folly::Optional())) + .WillOnce(Return(folly::unit)); EXPECT_CALL(transport_, sendWTMaxData(kDefaultWTReceiveWindow)).Times(0); auto fut = streamHandle.readHandle->readStreamData() @@ -1393,4 +1394,112 @@ TEST_F(HTTPTransactionWebTransportTest, FrequentFlowControlGrants) { .WillRepeatedly(Return(folly::unit)); } +// The handler closes the session, freeing the handle we just created. +// Returning it would leave deliverWTStream registering freed memory with quic. +TEST_F(HTTPTransactionWebTransportTest, CloseSessionFromOnNewUniStream) { + auto* wtImpl = dynamic_cast(wt_); + ASSERT_NE(wtImpl, nullptr); + auto* txn = txn_.get(); + // closeSession() may detach the txn, so keep it alive for the assertions. + HTTPTransaction::DestructorGuard dg(txn); + + EXPECT_CALL(handler_, onWebTransportUniStream(_, _)) + .WillOnce( + [this](HTTPCodec::StreamID, WebTransport::StreamReadHandle* handle) { + EXPECT_NE(handle, nullptr); + EXPECT_CALL(transport_, sendEOM(txn_.get(), nullptr)); + EXPECT_CALL(transport_, sendAbort(txn_.get(), _)); + wt_->closeSession(); + }); + EXPECT_CALL( + transport_, + stopReadingWebTransportIngress(2, makeOpt(WebTransport::kSessionGone))) + .WillRepeatedly(Return(folly::unit)); + + auto* handle = txn->onWebTransportUniStream(2); + + EXPECT_EQ(handle, nullptr); + EXPECT_EQ(wtImpl->getReadHandle(2), nullptr); + EXPECT_EQ(wtImpl->readStreamData(2).error(), + WebTransport::ErrorCode::INVALID_STREAM_ID); +} + +// Same hole on the bidi path. +TEST_F(HTTPTransactionWebTransportTest, CloseSessionFromOnNewBidiStream) { + auto* wtImpl = dynamic_cast(wt_); + ASSERT_NE(wtImpl, nullptr); + auto* txn = txn_.get(); + HTTPTransaction::DestructorGuard dg(txn); + + EXPECT_CALL(handler_, onWebTransportBidiStream(_, _)) + .WillOnce( + [this](HTTPCodec::StreamID, WebTransport::BidiStreamHandle bidi) { + EXPECT_NE(bidi.readHandle, nullptr); + EXPECT_CALL(transport_, sendEOM(txn_.get(), nullptr)); + EXPECT_CALL(transport_, sendAbort(txn_.get(), _)); + wt_->closeSession(); + }); + EXPECT_CALL( + transport_, + stopReadingWebTransportIngress(1, makeOpt(WebTransport::kSessionGone))) + .WillRepeatedly(Return(folly::unit)); + EXPECT_CALL(transport_, + resetWebTransportEgress(1, WebTransport::kSessionGone)) + .WillRepeatedly(Return(folly::unit)); + + auto bidi = txn->onWebTransportBidiStream(1); + + EXPECT_EQ(bidi.readHandle, nullptr); + EXPECT_EQ(bidi.writeHandle, nullptr); + EXPECT_EQ(wtImpl->getReadHandle(1), nullptr); +} + +// A stream that buffered data + FIN is no longer open(), but its read callback +// is still registered. Teardown has to unregister it before freeing it. +TEST_F(HTTPTransactionWebTransportTest, TerminateSessionUnregistersEofStream) { + WebTransport::StreamReadHandle* readHandle{nullptr}; + EXPECT_CALL(handler_, onWebTransportUniStream(_, _)) + .WillOnce(SaveArg<1>(&readHandle)); + + auto* implHandle = txn_->onWebTransportUniStream(2); + ASSERT_NE(implHandle, nullptr); + ASSERT_NE(readHandle, nullptr); + + // Data + FIN arrive with no readStreamData() outstanding. + EXPECT_CALL(transport_, readWebTransportData(2, 65535)) + .WillOnce([](auto, auto) { return std::make_pair(makeBuf(10), true); }); + implHandle->readAvailable(2); + EXPECT_FALSE(implHandle->open()); + + EXPECT_CALL(transport_, stopReadingWebTransportIngress(2, _)) + .Times(1) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(transport_, sendEOM(txn_.get(), nullptr)); + EXPECT_CALL(transport_, sendAbort(txn_.get(), _)); + wt_->closeSession(); +} + +// Aborting the txn leaves the stream in wtIngressStreams_, so the re-lookup +// finds a handle that releasing the guard frees a moment later. +TEST_F(HTTPTransactionWebTransportTest, AbortTxnFromOnNewUniStream) { + auto* txn = txn_.get(); + + EXPECT_CALL(handler_, onWebTransportUniStream(_, _)) + .WillOnce( + [this](HTTPCodec::StreamID, WebTransport::StreamReadHandle* handle) { + EXPECT_NE(handle, nullptr); + // Completes ingress and egress: the txn dies with the last guard. + EXPECT_CALL(transport_, sendAbort(txn_.get(), _)); + txn_->sendAbort(); + }); + EXPECT_CALL(transport_, stopReadingWebTransportIngress(2, _)) + .WillRepeatedly(Return(folly::unit)); + + auto* handle = txn->onWebTransportUniStream(2); + + // The txn, webTransportImpl_ and the handle are all gone by now. + EXPECT_EQ(txn_, nullptr); + EXPECT_EQ(handle, nullptr); +} + } // namespace proxygen::test diff --git a/proxygen/lib/http/webtransport/WebTransportImpl.cpp b/proxygen/lib/http/webtransport/WebTransportImpl.cpp index 7c75d39c6..00a80154d 100644 --- a/proxygen/lib/http/webtransport/WebTransportImpl.cpp +++ b/proxygen/lib/http/webtransport/WebTransportImpl.cpp @@ -40,12 +40,19 @@ void WebTransportImpl::terminateSessionStreams(uint32_t errorCode, for (auto& [id, stream] : wtIngressStreams) { // Deliver an error to the application if needed VLOG(4) << "aborting wt ingress id=" << id << " err=" << errorCode - << "; open=" << int(stream.open()); + << "; open=" << int(stream.open()) + << "; readCb=" << int(stream.readCallbackRegistered()); if (stream.open()) { stream.deliverReadError(WebTransport::Exception(errorCode, reason)); - stopReadingWebTransportIngress(id, errorCode); + } + // Not the same as open(): a handle that saw a FIN with no read + // outstanding is closed but still registered, and about to be destroyed. + if (stream.readCallbackRegistered()) { + stream.onReadCallbackUnregistered(); + tp_.stopReadingWebTransportIngress(id, errorCode); } } + sp_.refreshTimeout(); auto wtEgressStreams = std::move(wtEgressStreams_); for (auto& [id, stream] : wtEgressStreams) { @@ -217,6 +224,14 @@ folly::Expected WebTransportImpl::stopReadingWebTransportIngress( HTTPCodec::StreamID id, folly::Optional errorCode) { auto res = tp_.stopReadingWebTransportIngress(id, errorCode); + if (!res.hasError()) { + // On failure the transport may still hold the handle, so leave it marked + // registered and let teardown unregister it. + auto it = wtIngressStreams_.find(id); + if (it != wtIngressStreams_.end()) { + it->second.onReadCallbackUnregistered(); + } + } sp_.refreshTimeout(); return res; } @@ -429,16 +444,15 @@ WebTransportImpl::StreamReadHandle::readStreamData() { VLOG(4) << __func__ << " waiting for data"; auto contract = folly::makePromiseContract(); readPromise_ = std::move(contract.promise); - readPromise_.setInterruptHandler( - [this](const folly::exception_wrapper& ex) { - VLOG(4) << "Exception from interrupt handler ex=" << ex.what(); - CHECK(ex.with_exception([this](const folly::FutureCancellation& ex) { - // TODO: allow app to configure the reset code on cancellation? - impl_.tp_.stopReadingWebTransportIngress( - id_, WebTransport::kInternalError); - deliverReadError(ex); - })) << "Unexpected exception type"; - }); + readPromise_.setInterruptHandler([this]( + const folly::exception_wrapper& ex) { + VLOG(4) << "Exception from interrupt handler ex=" << ex.what(); + CHECK(ex.with_exception([this](const folly::FutureCancellation& ex) { + // TODO: allow app to configure the reset code on cancellation? + impl_.stopReadingWebTransportIngress(id_, WebTransport::kInternalError); + deliverReadError(ex); + })) << "Unexpected exception type"; + }); return std::move(contract.future); } else { VLOG(4) << __func__ << " returning data len=" << buf_.chainLength(); @@ -515,7 +529,8 @@ WebTransport::FCState WebTransportImpl::StreamReadHandle::dataAvailable( void WebTransportImpl::StreamReadHandle::readError( quic::StreamId id, quic::QuicError error) noexcept { - // Do I need to setReadCallback(id, nullptr); + // The transport delivers readError() and drops the read callback itself. + readCallbackRegistered_ = false; impl_.sp_.refreshTimeout(); auto quicAppErrorCode = error.code.asApplicationErrorCode(); if (quicAppErrorCode) { diff --git a/proxygen/lib/http/webtransport/WebTransportImpl.h b/proxygen/lib/http/webtransport/WebTransportImpl.h index 2ee848f6f..da1169e71 100644 --- a/proxygen/lib/http/webtransport/WebTransportImpl.h +++ b/proxygen/lib/http/webtransport/WebTransportImpl.h @@ -324,6 +324,15 @@ class WebTransportImpl : public WebTransport { return !eof_ && !ex_; } + // True while the transport holds this handle as the stream's read + // callback. Destroying the handle while set leaves a dangling callback. + [[nodiscard]] bool readCallbackRegistered() const { + return readCallbackRegistered_; + } + void onReadCallbackUnregistered() { + readCallbackRegistered_ = false; + } + // quic::StreamReadCallback overrides void readAvailable(quic::StreamId id) noexcept override; void readError(quic::StreamId id, quic::QuicError error) noexcept override; @@ -333,6 +342,7 @@ class WebTransportImpl : public WebTransport { folly::Promise readPromise_; folly::IOBufQueue buf_{folly::IOBufQueue::cacheChainLength()}; bool eof_{false}; + bool readCallbackRegistered_{true}; }; private: @@ -430,6 +440,15 @@ class WebTransportImpl : public WebTransport { void closeEgressStream(HTTPCodec::StreamID id); void closeIngressStream(HTTPCodec::StreamID id); + [[nodiscard]] StreamReadHandle* getReadHandle(HTTPCodec::StreamID id) { + auto it = wtIngressStreams_.find(id); + return it == wtIngressStreams_.end() ? nullptr : &it->second; + } + [[nodiscard]] StreamWriteHandle* getWriteHandle(HTTPCodec::StreamID id) { + auto it = wtEgressStreams_.find(id); + return it == wtEgressStreams_.end() ? nullptr : &it->second; + } + [[nodiscard]] bool isBidirectional(HTTPCodec::StreamID id) const { return (id & 0b10) == 0; }