From 3c2628f4692723c691d9b14dea090fbba291de85 Mon Sep 17 00:00:00 2001 From: Facebook GitHub Bot Date: Wed, 4 Nov 2026 15:35:27 -0800 Subject: [PATCH] Re-sync with internal repository 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: 5b88be121b1c25751cfa299d99aa40af0173fce5 --- proxygen/lib/http/codec/HQFramedCodec.cpp | 129 ++++++-- proxygen/lib/http/codec/HQFramedCodec.h | 11 + proxygen/lib/http/codec/HQStreamCodec.cpp | 18 ++ proxygen/lib/http/codec/HQStreamCodec.h | 1 + .../http/codec/compress/HPACKDecodeBuffer.cpp | 8 +- .../lib/http/codec/compress/QPACKDecoder.cpp | 21 +- .../codec/compress/test/QPACKContextTests.cpp | 24 ++ proxygen/lib/http/codec/test/HQCodecTest.cpp | 91 ++++++ proxygen/lib/http/coro/HTTPCoroSession.cpp | 13 +- .../test/HTTPDownstreamCoroSessionTests.cpp | 30 ++ proxygen/lib/http/session/HQSession.cpp | 42 +-- proxygen/lib/http/session/HQSession.h | 13 +- .../session/test/HQDownstreamSessionTest.cpp | 298 +++++++++++++++++- .../session/test/HQDownstreamSessionTest.h | 4 + .../http/session/test/HQSessionTestCommon.h | 9 +- .../session/test/HQUpstreamSessionTest.cpp | 35 ++ 16 files changed, 675 insertions(+), 72 deletions(-) diff --git a/proxygen/lib/http/codec/HQFramedCodec.cpp b/proxygen/lib/http/codec/HQFramedCodec.cpp index 13b35ad79c..c4ce917319 100644 --- a/proxygen/lib/http/codec/HQFramedCodec.cpp +++ b/proxygen/lib/http/codec/HQFramedCodec.cpp @@ -14,13 +14,68 @@ #include #include +#include #include +#include namespace proxygen::hq { using namespace folly::io; using namespace folly; +namespace { +// CANCEL_PUSH, GOAWAY and MAX_PUSH_ID carry a single varint. +constexpr uint64_t kMaxIdFrameLength = 8; +constexpr uint64_t kMaxControlFrameLength = 4096; +// Partial frames in longer chains are copied to keep rescans short. +constexpr size_t kMaxPendingChainElements = 10; +constexpr size_t kPendingPayloadAllocSize = 4096; + +// folly::none for frame types that are skipped. +folly::Optional maxFrameLength(FrameType type) { + switch (type) { + case FrameType::DATA: + case FrameType::HEADERS: + case FrameType::PUSH_PROMISE: + return std::numeric_limits::max(); + case FrameType::CANCEL_PUSH: + case FrameType::GOAWAY: + case FrameType::MAX_PUSH_ID: + return kMaxIdFrameLength; + case FrameType::SETTINGS: + case FrameType::PRIORITY_UPDATE: + case FrameType::PUSH_PRIORITY_UPDATE: + case FrameType::FB_PRIORITY_UPDATE: + case FrameType::FB_PUSH_PRIORITY_UPDATE: + return kMaxControlFrameLength; + case FrameType::PADDING: + case FrameType::WEBTRANSPORT_BIDI: + default: + return folly::none; + } +} +} // namespace + +ParseResult HQFramedCodec::checkFrameLength(const FrameHeader& header) { + if (header.length > maxFrameLength(header.type).value_or(0)) { + setParseErrorContext( + "frame-too-large", + folly::to("type=", getFrameTypeString(header.type))); + return HTTP3::ErrorCode::HTTP_FRAME_ERROR; + } + return folly::none; +} + +size_t HQFramedCodec::bufferPayload(Cursor& cursor, size_t bufLen) { + const uint64_t missing = curHeader_.length - pendingPayload_.chainLength(); + const size_t toCopy = std::min(missing, bufLen); + QueueAppender appender(&pendingPayload_, + std::min(missing, kPendingPayloadAllocSize)); + appender.push(cursor, toCopy); + cursor.skip(toCopy); + return toCopy; +} + ParseResult HQFramedCodec::parseFrame(Cursor& cursor) { switch (curHeader_.type) { case hq::FrameType::DATA: @@ -43,25 +98,15 @@ ParseResult HQFramedCodec::parseFrame(Cursor& cursor) { case hq::FrameType::FB_PUSH_PRIORITY_UPDATE: return parsePriorityUpdate(cursor, curHeader_); default: - // Implementations MUST ignore and discard any frame that has a - // type that is unknown - if (callback_) { - callback_->onUnknownFrame(streamId_, - static_cast(curHeader_.type)); - } - break; + LOG(DFATAL) << "Unparsed frame type=" << curHeader_.type; + cursor.skip(curHeader_.length); + return folly::none; } - - VLOG(3) << "Skipping frame (type=" << (uint64_t)curHeader_.type << ")"; - cursor.skip(curHeader_.length); - return folly::none; } size_t HQFramedCodec::onFramedIngress(const IOBuf& buf) { FOLLY_SCOPED_TRACE_SECTION("HQFramedCodec - onFramedIngress"); - // if for some reason onFramedIngress gets called again after erroring out, - // skip parsing - if (connError_ != folly::none) { + if (connError_ != folly::none || parserPaused_) { return 0; } Cursor cursor(&buf); @@ -111,40 +156,62 @@ size_t HQFramedCodec::onFramedIngress(const IOBuf& buf) { receivedFrameCount_++; #endif pendingDataFrameBytes_ = curHeader_.length; - if (curHeader_.length == 0) { + if (!maxFrameLength(curHeader_.type)) { + // Implementations MUST ignore and discard any frame that has a + // type that is unknown + if (callback_) { + callback_->onUnknownFrame(streamId_, + static_cast(curHeader_.type)); + } + frameState_ = curHeader_.length == 0 + ? FrameState::FRAME_HEADER_TYPE + : FrameState::FRAME_PAYLOAD_STREAMING; + } else if (curHeader_.length == 0) { // If the frame length is zero, call parseFrame immediately. It is up // to each frame to determine whether length 0 is valid. connError_ = parseFrame(cursor); frameState_ = FrameState::FRAME_HEADER_TYPE; + } else if (curHeader_.type == FrameType::DATA) { + frameState_ = FrameState::FRAME_PAYLOAD_STREAMING; } else { - // For DATA frames, move to the streaming state - if (curHeader_.type == FrameType::DATA) { - frameState_ = FrameState::FRAME_PAYLOAD_STREAMING; - } else { - frameState_ = FrameState::FRAME_PAYLOAD; - } + connError_ = checkFrameLength(curHeader_); + frameState_ = FrameState::FRAME_PAYLOAD; } } else if (frameState_ == FrameState::FRAME_PAYLOAD) { // Already parsed the common frame header const auto frameLen = curHeader_.length; - if (bufLen >= frameLen) { + if (pendingPayload_.empty() && bufLen >= frameLen) { connError_ = parseFrame(cursor); // if connError_ is set, it means there was a frame error, // so it doesn't really matter whether we have actually parsed all the // data or not parsed += curHeader_.length; frameState_ = FrameState::FRAME_HEADER_TYPE; - } else { + } else if (pendingPayload_.empty() && + buf.countChainElements() <= kMaxPendingChainElements) { break; + } else { + parsed += bufferPayload(cursor, bufLen); + if (pendingPayload_.chainLength() == frameLen) { + auto payload = pendingPayload_.move(); + Cursor payloadCursor(payload.get()); + connError_ = parseFrame(payloadCursor); + frameState_ = FrameState::FRAME_HEADER_TYPE; + } } } else if (bufLen > 0 && frameState_ == FrameState::FRAME_PAYLOAD_STREAMING) { - FrameHeader auxDataFrameHeader; - auxDataFrameHeader.type = FrameType::DATA; - auxDataFrameHeader.length = std::min(pendingDataFrameBytes_, bufLen); - connError_ = parseData(cursor, auxDataFrameHeader); - parsed += auxDataFrameHeader.length; - pendingDataFrameBytes_ -= auxDataFrameHeader.length; + const auto chunkLen = std::min(pendingDataFrameBytes_, bufLen); + if (curHeader_.type == FrameType::DATA) { + FrameHeader auxDataFrameHeader; + auxDataFrameHeader.type = FrameType::DATA; + auxDataFrameHeader.length = chunkLen; + connError_ = parseData(cursor, auxDataFrameHeader); + } else { + cursor.skip(chunkLen); + } + parsed += chunkLen; + pendingDataFrameBytes_ -= chunkLen; if (pendingDataFrameBytes_ == 0) { frameState_ = FrameState::FRAME_HEADER_TYPE; } @@ -154,7 +221,9 @@ size_t HQFramedCodec::onFramedIngress(const IOBuf& buf) { parsedTot += parsed; totalBytesParsed_ += parsed; } - checkConnectionError(connError_, &buf); + if (!checkConnectionError(connError_, &buf) && parserPaused_ && pauseHook_) { + pauseHook_(); + } return parsedTot; } diff --git a/proxygen/lib/http/codec/HQFramedCodec.h b/proxygen/lib/http/codec/HQFramedCodec.h index bc37e0ff2c..bd7af6c8ed 100644 --- a/proxygen/lib/http/codec/HQFramedCodec.h +++ b/proxygen/lib/http/codec/HQFramedCodec.h @@ -11,6 +11,7 @@ #include #include +#include #include namespace proxygen::hq { @@ -71,6 +72,11 @@ class HQFramedCodec : public HTTPCodec { resumeHook_ = std::move(resumeHook); } + // Invoked when ingress parsing stops with the parser paused. + void setPauseHook(folly::Function pauseHook) { + pauseHook_ = std::move(pauseHook); + } + bool isParserPaused() const override { return parserPaused_; } @@ -265,6 +271,7 @@ class HQFramedCodec : public HTTPCodec { } virtual ParseResult checkFrameAllowed(FrameType type) = 0; + virtual ParseResult checkFrameLength(const FrameHeader& header); virtual ParseResult parseData(folly::io::Cursor& /*cursor*/, const FrameHeader& /*header*/) { @@ -337,8 +344,11 @@ class HQFramedCodec : public HTTPCodec { ParseResult parseFrame(folly::io::Cursor& cursor); bool checkConnectionError(ParseResult err, const folly::IOBuf* buf); + size_t bufferPayload(folly::io::Cursor& cursor, size_t bufLen); + // Current frame state size_t pendingDataFrameBytes_{0}; + folly::IOBufQueue pendingPayload_{folly::IOBufQueue::cacheChainLength()}; #ifndef NDEBUG uint64_t receivedFrameCount_{0}; @@ -356,6 +366,7 @@ class HQFramedCodec : public HTTPCodec { std::string parseErrorAdditionalInfo_; uint64_t totalBytesParsed_{0}; folly::Function resumeHook_; + folly::Function pauseHook_; }; } // namespace proxygen::hq diff --git a/proxygen/lib/http/codec/HQStreamCodec.cpp b/proxygen/lib/http/codec/HQStreamCodec.cpp index d613a8313d..bc9d7a7f82 100644 --- a/proxygen/lib/http/codec/HQStreamCodec.cpp +++ b/proxygen/lib/http/codec/HQStreamCodec.cpp @@ -16,6 +16,8 @@ #include #include +#include + namespace proxygen::hq { using namespace folly; @@ -101,6 +103,22 @@ ParseResult HQStreamCodec::checkFrameAllowed(FrameType type) { return folly::none; } +ParseResult HQStreamCodec::checkFrameLength(const FrameHeader& header) { + if (header.type != hq::FrameType::HEADERS && + header.type != hq::FrameType::PUSH_PROMISE) { + return HQFramedCodec::checkFrameLength(header); + } + if (header.length > headerCodec_.getMaxUncompressed()) { + if (header.type == hq::FrameType::HEADERS && !finalIngressHeadersSeen_ && + callback_) { + callback_->onMessageBegin(streamId_, nullptr); + } + setParserPaused(true); + onDecodeError(HPACK::DecodeError::HEADERS_TOO_LARGE); + } + return folly::none; +} + ParseResult HQStreamCodec::parseData(Cursor& cursor, const FrameHeader& header) { // NOTE: If an error path is added to this method, it needs to setParserPaused diff --git a/proxygen/lib/http/codec/HQStreamCodec.h b/proxygen/lib/http/codec/HQStreamCodec.h index 091a24d372..a1e6b8760d 100644 --- a/proxygen/lib/http/codec/HQStreamCodec.h +++ b/proxygen/lib/http/codec/HQStreamCodec.h @@ -159,6 +159,7 @@ class HQStreamCodec protected: ParseResult checkFrameAllowed(FrameType type) override; + ParseResult checkFrameLength(const FrameHeader& header) override; ParseResult parseData(folly::io::Cursor& cursor, const FrameHeader& header) override; ParseResult parseHeaders(folly::io::Cursor& cursor, diff --git a/proxygen/lib/http/codec/compress/HPACKDecodeBuffer.cpp b/proxygen/lib/http/codec/compress/HPACKDecodeBuffer.cpp index 82cbff9f67..f0fa5741df 100644 --- a/proxygen/lib/http/codec/compress/HPACKDecodeBuffer.cpp +++ b/proxygen/lib/http/codec/compress/HPACKDecodeBuffer.cpp @@ -69,15 +69,15 @@ DecodeError HPACKDecodeBuffer::decodeLiteral(uint8_t nbit, EOB_LOG("Could not decode literal size", result); return result; } + if (size > maxLiteralSize_) { + LOG(ERROR) << "Literal too large, size=" << size; + return DecodeError::LITERAL_TOO_LARGE; + } if (size > remainingBytes_) { EOB_LOG(folly::to( "size(", size, ") > remainingBytes_(", remainingBytes_, ")")); return DecodeError::BUFFER_UNDERFLOW; } - if (size > maxLiteralSize_) { - LOG(ERROR) << "Literal too large, size=" << size; - return DecodeError::LITERAL_TOO_LARGE; - } const uint8_t* data; unique_ptr tmpbuf; // handle the case where the buffer spans multiple buffers diff --git a/proxygen/lib/http/codec/compress/QPACKDecoder.cpp b/proxygen/lib/http/codec/compress/QPACKDecoder.cpp index 639c26b649..d05b904afc 100644 --- a/proxygen/lib/http/codec/compress/QPACKDecoder.cpp +++ b/proxygen/lib/http/codec/compress/QPACKDecoder.cpp @@ -10,6 +10,9 @@ #include +#include +#include + using folly::io::Cursor; namespace { @@ -201,10 +204,14 @@ HPACK::DecodeError QPACKDecoder::decodeEncoderStream( std::unique_ptr buf) { ingress_.append(std::move(buf)); Cursor cursor(ingress_.front()); - HPACKDecodeBuffer dbuf(cursor, - ingress_.chainLength(), - maxUncompressed_, - /* endOfBufferIsError=*/false); + // Every inserted literal must fit in the dynamic table. A Huffman code is at + // most 30 bits, so the encoded form can be up to 4x the decoded size. + HPACKDecodeBuffer dbuf( + cursor, + ingress_.chainLength(), + uint32_t(std::min(uint64_t(maxTableSize_) * 4, + std::numeric_limits::max())), + /* endOfBufferIsError=*/false); VLOG(6) << "Decoding control block"; baseIndex_ = 0; @@ -321,6 +328,12 @@ uint32_t QPACKDecoder::decodeLiteralHeaderQ( LOG(ERROR) << "Error decoding header name err_=" << err_; return 0; } + if (indexing && + headerName.size() + HPACKHeader::kMinLength > table_.capacity()) { + LOG(ERROR) << "Encoder inserted a name larger than capacity"; + err_ = HPACK::DecodeError::INSERT_TOO_LARGE; + return 0; + } partial->header.name = HPACKHeaderName{headerName}; } partial->state = Partial::VALUE; diff --git a/proxygen/lib/http/codec/compress/test/QPACKContextTests.cpp b/proxygen/lib/http/codec/compress/test/QPACKContextTests.cpp index 0ecbb21866..77e1ec0eb0 100644 --- a/proxygen/lib/http/codec/compress/test/QPACKContextTests.cpp +++ b/proxygen/lib/http/codec/compress/test/QPACKContextTests.cpp @@ -909,6 +909,30 @@ TEST(QPACKContextTests, RejectsOverflowingRequiredInsertCount) { decoder, headerBlock.release(), HPACK::DecodeError::INVALID_INDEX); } +TEST(QPACKContextTests, EncoderStreamLiteralLargerThanTable) { + QPACKDecoder decoder(HPACK::kTableSize); + HPACKEncodeBuffer partial(128, false); + partial.encodeInteger(100, HPACK::Q_INSERT_NO_NAME_REF); + EXPECT_EQ(decoder.decodeEncoderStream(partial.release()), + HPACK::DecodeError::NONE); + + QPACKDecoder decoder2(HPACK::kTableSize); + HPACKEncodeBuffer tooLarge(128, false); + tooLarge.encodeInteger(4 * HPACK::kTableSize + 1, + HPACK::Q_INSERT_NO_NAME_REF); + EXPECT_EQ(decoder2.decodeEncoderStream(tooLarge.release()), + HPACK::DecodeError::LITERAL_TOO_LARGE); + + QPACKDecoder decoder3(HPACK::kTableSize); + HPACKEncodeBuffer bigName(128, false); + bigName.encodeLiteral( + HPACK::Q_INSERT_NO_NAME_REF.code, + HPACK::Q_INSERT_NO_NAME_REF.prefixLength, + string(HPACK::kTableSize - HPACKHeader::kMinLength + 1, 'x')); + EXPECT_EQ(decoder3.decodeEncoderStream(bigName.release()), + HPACK::DecodeError::INSERT_TOO_LARGE); +} + TEST(QPACKContextTests, DecodeErrors) { QPACKDecoder decoder(128); unique_ptr buf = IOBuf::create(128); diff --git a/proxygen/lib/http/codec/test/HQCodecTest.cpp b/proxygen/lib/http/codec/test/HQCodecTest.cpp index c85e1b28d0..71d2eee512 100644 --- a/proxygen/lib/http/codec/test/HQCodecTest.cpp +++ b/proxygen/lib/http/codec/test/HQCodecTest.cpp @@ -372,16 +372,24 @@ void HQCodecTestFixture::qpackTest(bool blocked) { } TestStreamingCallback cb; CHECK(!queue_.empty()); + size_t pauses = 0; + downstreamCodec_->setPauseHook([&] { + EXPECT_TRUE(downstreamCodec_->isParserPaused()); + ++pauses; + }); upstreamCodec_->generateBody( queue_, streamId, folly::IOBuf::copyBuffer("ohai"), 0, false); upstreamCodec_->generateEOM(queue_, streamId); auto consumed = downstreamCodec_->onIngress(*queue_.front()); + EXPECT_EQ(pauses, blocked ? 1 : 0); EXPECT_TRUE(blocked || consumed == queue_.chainLength()); queue_.trimStart(consumed); EXPECT_EQ(callbacks_.streamErrors, 0); EXPECT_EQ(callbacks_.sessionErrors, 0); if (blocked) { EXPECT_EQ(callbacks_.headersComplete, 0); + EXPECT_EQ(downstreamCodec_->onIngress(*queue_.front()), 0); + EXPECT_EQ(pauses, 1); qpackEncoderCodec_.onUnidirectionalIngress(qpackUpEncoderWriteBuf_.move()); EXPECT_EQ(callbacks_.sessionErrors, 0); } @@ -1037,6 +1045,89 @@ TEST_F(HQCodecTest, PaddingIgnored) { EXPECT_EQ(callbacks_.lastParseError, nullptr); } +TEST_F(HQCodecTest, FragmentedControlFramesBoundUnparsedChain) { + IOBufQueue input{IOBufQueue::cacheChainLength()}; + writeValidFrame(input, FrameType::SETTINGS); + writeFrameHeaderManual(input, 0x21, 100); + input.append(std::string(100, 'x')); + hq::writePriorityUpdate( + input, + 0, + folly::to("u=1, x=\"", std::string(100, 'a'), "\"")); + writeValidFrame(input, FrameType::GOAWAY); + + while (!input.empty()) { + queueCtrl_.append(input.split(1)); + parseControl(CodecType::CONTROL_DOWNSTREAM); + EXPECT_LE(queueCtrl_.empty() ? 0 : queueCtrl_.front()->countChainElements(), + 10); + } + EXPECT_EQ(callbacks_.settings, 1); + EXPECT_EQ(callbacks_.unknownFrames, 1); + EXPECT_EQ(callbacks_.urgency, 1); + EXPECT_EQ(callbacks_.goaways, 1); + EXPECT_EQ(callbacks_.sessionErrors, 0); +} + +TEST_F(HQCodecTest, ControlFrameTooLarge) { + writeFrameHeaderManual( + queueCtrl_, static_cast(FrameType::SETTINGS), 4097); + parseControl(CodecType::CONTROL_DOWNSTREAM); + EXPECT_EQ(callbacks_.sessionErrors, 1); + EXPECT_EQ(callbacks_.lastParseError->getHttp3ErrorCode(), + HTTP3::ErrorCode::HTTP_FRAME_ERROR); +} + +TEST_F(HQCodecTest, IdFrameTooLarge) { + writeValidFrame(queueCtrl_, FrameType::SETTINGS); + writeFrameHeaderManual( + queueCtrl_, static_cast(FrameType::GOAWAY), 9); + parseControl(CodecType::CONTROL_DOWNSTREAM); + EXPECT_EQ(callbacks_.sessionErrors, 1); + EXPECT_EQ(callbacks_.lastParseError->getHttp3ErrorCode(), + HTTP3::ErrorCode::HTTP_FRAME_ERROR); +} + +TEST_F(HQCodecTest, FragmentedHeadersThenDataInFinalChunk) { + // Feed HEADERS in small chunks so the payload is copied into the codec, then + // deliver the end of HEADERS and a DATA frame in one chunk. + IOBufQueue input{IOBufQueue::cacheChainLength()}; + HTTPMessage req = getPostRequest(10); + req.getHeaders().add("x-pad", std::string(200, 'a')); + upstreamCodec_->generateHeader(input, streamId_, req, false); + const auto headersLen = input.chainLength(); + upstreamCodec_->generateBody( + input, streamId_, makeBuf(10), HTTPCodec::NoPadding, true); + + constexpr size_t kChunk = 7; + for (size_t sent = 0; headersLen - sent > kChunk; sent += kChunk) { + queue_.append(input.split(kChunk)); + parse(); + EXPECT_LE(queue_.empty() ? 0 : queue_.front()->countChainElements(), 10); + } + queue_.append(input.move()); + parse(); + downstreamCodec_->onIngressEOF(); + + EXPECT_TRUE(queue_.empty()); + EXPECT_EQ(callbacks_.headersComplete, 1); + EXPECT_EQ(callbacks_.bodyLength, 10); + EXPECT_EQ(callbacks_.messageComplete, 1); + EXPECT_EQ(callbacks_.streamErrors, 0); + EXPECT_EQ(callbacks_.sessionErrors, 0); +} + +TEST_F(HQCodecTest, HeadersFrameLargerThanHeaderListLimit) { + qpackDownstream_.setMaxUncompressed(37); + writeFrameHeaderManual(queue_, static_cast(FrameType::HEADERS), 38); + parse(); + EXPECT_EQ(callbacks_.messageBegin, 1); + EXPECT_EQ(callbacks_.streamErrors, 1); + EXPECT_EQ(callbacks_.sessionErrors, 0); + EXPECT_EQ(callbacks_.lastParseError->getHttp3ErrorCode(), + HTTP3::ErrorCode::HTTP_QPACK_DECOMPRESSION_FAILED); +} + struct FrameAllowedParams { CodecType codecType; FrameType frameType; diff --git a/proxygen/lib/http/coro/HTTPCoroSession.cpp b/proxygen/lib/http/coro/HTTPCoroSession.cpp index 03259d8cf3..3e9321496b 100644 --- a/proxygen/lib/http/coro/HTTPCoroSession.cpp +++ b/proxygen/lib/http/coro/HTTPCoroSession.cpp @@ -3170,11 +3170,16 @@ void HTTPCoroSession::handleWriteEventTimeout() { void HTTPCoroSession::resetOpenStreams(HTTPErrorCode error, std::string_view details) { // Reset all open streams - std::vector streams; - for (auto& [_, stream] : streams_) { - streams.push_back(stream.get()); + std::vector ids; + ids.reserve(streams_.size()); + for (auto& [id, _] : streams_) { + ids.push_back(id); } - for (auto* stream : streams) { + for (auto id : ids) { + auto* stream = findStream(id); + if (!stream) { + continue; + } // Queue an ingress abort for any stream that hasn't seen EOM yet stream->abortIngress(error, details); // marks egress complete, clears queued egress, checks for detach diff --git a/proxygen/lib/http/coro/test/HTTPDownstreamCoroSessionTests.cpp b/proxygen/lib/http/coro/test/HTTPDownstreamCoroSessionTests.cpp index 25da97d9ba..f96e697f0c 100644 --- a/proxygen/lib/http/coro/test/HTTPDownstreamCoroSessionTests.cpp +++ b/proxygen/lib/http/coro/test/HTTPDownstreamCoroSessionTests.cpp @@ -3231,6 +3231,36 @@ TEST_P(H1DownstreamSessionTest, BodyReadTimeout) { parseOutput(); } +TEST_P(H1DownstreamSessionTest, PipelinedReadTimeoutThenWriteTimeout) { + InSequence enforceOrder; + session_->setStreamReadTimeout(std::chrono::milliseconds(100)); + session_->setWriteTimeout(std::chrono::milliseconds(250)); + auto id = sendRequestHeader(getGetRequest(), true); + writeBuf_.append(folly::IOBuf::copyBuffer("GET / HTTP/1.1")); + transport_->addReadEvent(id, writeBuf_.move(), false); + + auto handler1 = + addSimpleStrictHandler([this](folly::EventBase *evb, + HTTPSessionContextPtr /*ctx*/, + HTTPSourceHolder requestSource) + -> folly::coro::Task { + auto streamID = + co_await expectRequest(requestSource, HTTPMethod::GET, "/"); + transport_->pauseWrites(streamID); + co_return HTTPFixedSource::makeFixedResponse(200, makeBuf(100000)); + }); + auto handler2 = + addSimpleStrictHandler([](folly::EventBase *evb, + HTTPSessionContextPtr /*ctx*/, + HTTPSourceHolder requestSource) + -> folly::coro::Task { + co_await expectHeaderError(requestSource, HTTPErrorCode::READ_TIMEOUT); + co_return HTTPFixedSource::makeFixedResponse(408, makeBuf(100)); + }); + evb_.loop(); + expectedError_ = TransportErrorCode::NETWORK_ERROR; +} + TEST_P(HQDownstreamSessionTest, HeadersReadTimeout) { // Send an incomplete GET request. session_->setStreamReadTimeout(std::chrono::milliseconds(100)); diff --git a/proxygen/lib/http/session/HQSession.cpp b/proxygen/lib/http/session/HQSession.cpp index 4eb68141ec..961fa7e984 100644 --- a/proxygen/lib/http/session/HQSession.cpp +++ b/proxygen/lib/http/session/HQSession.cpp @@ -1223,7 +1223,11 @@ void HQSession::scheduleLoopCallback(bool thisIteration) { } } -void HQSession::resumeReads(quic::StreamId streamId) { +void HQSession::resumeReads(HQStreamTransportBase& stream) { + if (!sock_ || stream.readPausedForCodec_ || ingressLimitExceeded()) { + return; + } + auto streamId = stream.getIngressStreamId(); VLOG(4) << __func__ << " sess=" << *this << ": resuming reads id=" << streamId; sock_->resumeRead(streamId); @@ -1234,11 +1238,8 @@ void HQSession::resumeReads(quic::StreamId streamId) { void HQSession::resumeReads() { VLOG(4) << __func__ << " sess=" << *this << ": resuming reads"; - invokeOnIngressStreams([this](HQStreamTransportBase* hqStream) { - if (sock_) { - sock_->resumeRead(hqStream->getIngressStreamId()); - } - }); + invokeOnIngressStreams( + [this](HQStreamTransportBase* hqStream) { resumeReads(*hqStream); }); } void HQSession::pauseReads(quic::StreamId streamId) { @@ -2308,6 +2309,20 @@ void HQSession::HQStreamTransportBase::initCodec( auto c = dynamic_cast(realCodec_.get()); CHECK(c) << "HQ should use HQStream codec"; c->setActivationHook([this] { return setActiveCodec("self"); }); + c->setPauseHook([this] { + readPausedForCodec_ = true; + if (session_.sock_) { + session_.pauseReads(getIngressStreamId()); + } + }); + c->setResumeHook([this] { + if (readPausedForCodec_) { + readPausedForCodec_ = false; + session_.pendingProcessReadSet_.insert(getIngressStreamId()); + session_.resumeReads(*this); + session_.scheduleLoopCallback(); + } + }); hqStreamCodec_ = c; } auto g = folly::makeGuard(setActiveCodec(__func__)); @@ -2636,17 +2651,6 @@ void HQSession::HQStreamTransportBase::onHeadersComplete( // if we are getting a response, transportReady has been called! session_.headersComplete(msg.get()); - // onHeadersComplete can be triggered by data from a different stream ID - // - specifically, the QPACK encoder stream. If that's true, then there may - // be unparsed data in HQStreamTransport. Add this stream's id to the - // read set and schedule a loop callback to restart it. - if (session_.pendingProcessReadSet_.find(getStreamId()) == - session_.pendingProcessReadSet_.end() && - !readBuf_.empty()) { - session_.pendingProcessReadSet_.insert(getStreamId()); - session_.scheduleLoopCallback(); - } - auto timeDiff = std::chrono::duration_cast( std::chrono::steady_clock::now() - createdTime); auto sock = session_.sock_; @@ -3062,9 +3066,9 @@ void HQSession::HQStreamTransportBase::onError(HTTPCodec::StreamID streamID, return; } - if (!codecStreamId_ && error.hasHttpStatusCode() && streamID != 0) { + if (!codecStreamId_) { // onError before onMessageBegin - codecStreamId_ = streamID; + codecStreamId_ = codecFilterChain.createStream(); } if (!txn_.getHandler() && diff --git a/proxygen/lib/http/session/HQSession.h b/proxygen/lib/http/session/HQSession.h index 5f1af213f8..4d45c2c528 100644 --- a/proxygen/lib/http/session/HQSession.h +++ b/proxygen/lib/http/session/HQSession.h @@ -587,11 +587,6 @@ class HQSession // has been erased bool eraseStream(quic::StreamId); - void resumeReadsForPushStream(quic::StreamId streamId) { - pendingProcessReadSet_.insert(streamId); - resumeReads(streamId); - } - // Find a control stream by type HQControlStream* findControlStream(hq::UnidirectionalStreamType streamType); @@ -789,13 +784,13 @@ class HQSession // during the last event loop void processReadData(); - // Pausing reads prevents the read callback to be invoked on the stream - void resumeReads(quic::StreamId id); + // Resume a stream unless its codec or the session buffer limit blocks reads. + void resumeReads(HQStreamTransportBase& stream); // Resume all ingress transactions void resumeReads(); - // Resuming the reads allows the read callback to be involved + // Pausing reads prevents the read callback from being invoked on the stream. void pauseReads(quic::StreamId id); // Pause all ingress transactions @@ -1637,6 +1632,8 @@ class HQSession } // have read EOF bool readEOF_{false}; + // Tracks a parser pause that persists after onIngress returns. + bool readPausedForCodec_{false}; bool hasCodec_{false}; // The stream's HQ codec, when it has one; it reports how each generated // body was framed. diff --git a/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp b/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp index 522cf060d6..7bb322695c 100644 --- a/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp +++ b/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp @@ -2278,6 +2278,234 @@ TEST_P(HQDownstreamSessionTest, DelayedQPACK) { hqSession_->closeWhenIdle(); } +void HQDownstreamSessionTest::testQPACKBlockedReads(size_t initialBodyBytes) { + constexpr size_t kReadLimit = 16 * 1024; + constexpr size_t kChunkSize = 4096; + constexpr size_t kNumChunks = 64; + auto* session = static_cast(hqSession_); + session->setReadBufferLimit(kReadLimit); + + const auto numChunks = kNumChunks + (initialBodyBytes > 0 ? 1 : 0); + const auto totalBodyBytes = kChunkSize * kNumChunks + initialBodyBytes; + auto req = getPostRequest(totalBodyBytes); + req.getHeaders().add("X-FB-Debug", "rfccffgvtvnenjkbtitkfdufddnvbecu"); + auto id = sendRequest(req, false); + auto& request = getStream(id); + size_t bufferedBytes = 0; + if (initialBodyBytes > 0) { + bufferedBytes = request.codec->generateBody(request.buf, + request.id, + makeBuf(initialBodyBytes), + HTTPCodec::NoPadding, + false); + } + auto encoderInstructions = encoderWriteBuf_.move(); + ASSERT_NE(encoderInstructions, nullptr); + flushRequestsAndLoopN(1); + + auto* stream = session->findStream(id); + ASSERT_NE(stream, nullptr); + ASSERT_TRUE(stream->getCodec().isParserPaused()); + EXPECT_EQ(stream->readBuf_.chainLength(), bufferedBytes); + if (initialBodyBytes > kReadLimit) { + EXPECT_GT(stream->readBuf_.chainLength(), kReadLimit); + } + EXPECT_TRUE(socketDriver_->isStreamPaused(id)); + EXPECT_CALL(infoCb_, onIngressLimitExceeded(_)).Times(0); + + const auto initialBytesRead = socketDriver_->streams_[id].readOffset; + size_t queuedBytes = 0; + for (size_t i = 0; i < kNumChunks; ++i) { + SCOPED_TRACE(i); + request.codec->generateBody(request.buf, + request.id, + makeBuf(kChunkSize), + HTTPCodec::NoPadding, + false); + queuedBytes += request.buf.chainLength(); + flushRequestsAndLoopN(1); + + EXPECT_TRUE(stream->getCodec().isParserPaused()); + EXPECT_EQ(stream->readBuf_.chainLength(), bufferedBytes); + EXPECT_FALSE(session->ingressLimitExceeded()); + const auto& socketStream = socketDriver_->streams_[id]; + EXPECT_EQ(socketStream.readState, MockQuicSocketDriver::PAUSED); + EXPECT_FALSE(socketStream.error.hasValue()); + EXPECT_EQ(socketStream.readBuf.chainLength(), queuedBytes); + EXPECT_EQ(socketStream.readOffset, initialBytesRead); + } + auto handler = addSimpleStrictHandler(); + handler->expectHeaders(); + size_t bodyBytes = 0; + EXPECT_CALL(*handler, _onBodyWithOffset(_, _)) + .Times(numChunks) + .WillRepeatedly([&](uint64_t offset, std::shared_ptr body) { + EXPECT_EQ(offset, bodyBytes); + bodyBytes += body->computeChainDataLength(); + }); + encoderWriteBuf_.append(std::move(encoderInstructions)); + flushRequestsAndLoopN(1); + for (size_t i = 0; i < 10 && bodyBytes < totalBodyBytes; ++i) { + eventBase_.loopOnce(EVLOOP_NONBLOCK); + } + EXPECT_FALSE(stream->getCodec().isParserPaused()); + EXPECT_TRUE(stream->readBuf_.empty()); + EXPECT_FALSE(socketDriver_->isStreamPaused(id)); + EXPECT_EQ(bodyBytes, totalBodyBytes); + + handler->expectEOM([&] { handler->sendReplyWithBody(200, 100); }); + handler->expectDetachTransaction(); + request.readEOF = true; + flushRequestsAndLoop(); + hqSession_->closeWhenIdle(); +} + +TEST_P(HQDownstreamSessionTest, QPACKBlockedReads) { + testQPACKBlockedReads(0); +} + +TEST_P(HQDownstreamSessionTest, QPACKBlockedReadsWithBufferedBody) { + testQPACKBlockedReads(4096); +} + +TEST_P(HQDownstreamSessionTest, QPACKBlockedReadsWithLargeInitialBody) { + testQPACKBlockedReads(32 * 1024); +} + +void HQDownstreamSessionTest::testQPACKBlockedReadsWithSessionLimit( + bool unblockQPACKFirst) { + constexpr size_t kReadLimit = 4096; + auto* session = static_cast(hqSession_); + session->setReadBufferLimit(kReadLimit); + + auto id = sendRequest(getPostRequest(kReadLimit + 1), false); + auto handler = addSimpleStrictHandler(); + handler->expectHeaders([&] { handler->txn_->pauseIngress(); }); + flushRequestsAndLoopN(1); + + auto req = getPostRequest(10); + req.getHeaders().add("X-FB-Debug", "rfccffgvtvnenjkbtitkfdufddnvbecu"); + auto blockedId = sendRequest(req, false); + auto encoderInstructions = encoderWriteBuf_.move(); + ASSERT_NE(encoderInstructions, nullptr); + flushRequestsAndLoopN(1); + auto* blockedStream = session->findStream(blockedId); + ASSERT_NE(blockedStream, nullptr); + ASSERT_TRUE(blockedStream->getCodec().isParserPaused()); + EXPECT_TRUE(socketDriver_->isStreamPaused(blockedId)); + + auto& request = getStream(id); + request.codec->generateBody(request.buf, + request.id, + makeBuf(kReadLimit + 1), + HTTPCodec::NoPadding, + false); + flushRequestsAndLoopN(1); + EXPECT_TRUE(session->ingressLimitExceeded()); + EXPECT_TRUE(socketDriver_->isStreamPaused(id)); + + auto& blockedRequest = getStream(blockedId); + blockedRequest.codec->generateBody(blockedRequest.buf, + blockedRequest.id, + makeBuf(10), + HTTPCodec::NoPadding, + false); + flushRequestsAndLoopN(1); + const auto queuedBytes = + socketDriver_->streams_[blockedId].readBuf.chainLength(); + EXPECT_GT(queuedBytes, 0); + + auto blockedHandler = addSimpleStrictHandler(); + blockedHandler->expectHeaders(); + blockedHandler->expectBody(); + handler->expectBody(); + auto unblockQPACK = [&] { + encoderWriteBuf_.append(std::move(encoderInstructions)); + flushRequestsAndLoopN(1); + }; + if (unblockQPACKFirst) { + unblockQPACK(); + EXPECT_FALSE(blockedStream->getCodec().isParserPaused()); + EXPECT_TRUE(session->ingressLimitExceeded()); + } else { + handler->txn_->resumeIngress(); + EXPECT_TRUE(blockedStream->getCodec().isParserPaused()); + EXPECT_FALSE(session->ingressLimitExceeded()); + } + EXPECT_TRUE(socketDriver_->isStreamPaused(blockedId)); + EXPECT_TRUE(blockedStream->readBuf_.empty()); + EXPECT_EQ(socketDriver_->streams_[blockedId].readBuf.chainLength(), + queuedBytes); + + if (unblockQPACKFirst) { + handler->txn_->resumeIngress(); + } else { + unblockQPACK(); + } + for (size_t i = 0; + i < 10 && (socketDriver_->isStreamPaused(blockedId) || + !socketDriver_->streams_[blockedId].readBuf.empty() || + !blockedStream->readBuf_.empty()); + ++i) { + eventBase_.loopOnce(EVLOOP_NONBLOCK); + } + EXPECT_FALSE(socketDriver_->isStreamPaused(blockedId)); + EXPECT_TRUE(socketDriver_->streams_[blockedId].readBuf.empty()); + EXPECT_TRUE(blockedStream->readBuf_.empty()); + + handler->expectEOM([&] { handler->sendReplyWithBody(200, 100); }); + handler->expectDetachTransaction(); + blockedHandler->expectEOM( + [&] { blockedHandler->sendReplyWithBody(200, 100); }); + blockedHandler->expectDetachTransaction(); + request.readEOF = true; + blockedRequest.readEOF = true; + flushRequestsAndLoop(); + hqSession_->closeWhenIdle(); +} + +TEST_P(HQDownstreamSessionTest, QPACKBlockedReadsSessionLimitClearedFirst) { + testQPACKBlockedReadsWithSessionLimit(false); +} + +TEST_P(HQDownstreamSessionTest, QPACKBlockedReadsQPACKUnblockedFirst) { + testQPACKBlockedReadsWithSessionLimit(true); +} + +TEST_P(HQDownstreamSessionTest, QPACKUnblockedHeadersAbortClearsBufferedBody) { + auto req = getPostRequest(4096); + req.getHeaders().add("X-FB-Debug", "rfccffgvtvnenjkbtitkfdufddnvbecu"); + auto id = sendRequest(req, false); + auto& request = getStream(id); + request.codec->generateBody( + request.buf, request.id, makeBuf(4096), HTTPCodec::NoPadding, false); + auto encoderInstructions = encoderWriteBuf_.move(); + ASSERT_NE(encoderInstructions, nullptr); + flushRequestsAndLoopN(1); + + auto* session = static_cast(hqSession_); + auto* stream = session->findStream(id); + ASSERT_NE(stream, nullptr); + ASSERT_FALSE(stream->readBuf_.empty()); + ASSERT_TRUE(socketDriver_->isStreamPaused(id)); + + auto handler = addSimpleStrictHandler(); + handler->expectHeaders([&] { handler->txn_->sendAbort(); }); + handler->expectDetachTransaction(); + encoderWriteBuf_.append(std::move(encoderInstructions)); + flushRequestsAndLoopN(1); + for (size_t i = 0; i < 10 && session->getNumStreams() != 0; ++i) { + eventBase_.loopOnce(EVLOOP_NONBLOCK); + } + EXPECT_EQ(session->getNumStreams(), 0); + EXPECT_EQ(session->findStream(id), nullptr); + if (session->getNumStreams() != 0) { + session->dropConnection(); + return; + } + session->closeWhenIdle(); +} + TEST_P(HQDownstreamSessionTest, cancelQPACK) { auto req = getGetRequest(); req.getHeaders().add("X-FB-Debug", "rfccffgvtvnenjkbtitkfdufddnvbecu"); @@ -2300,21 +2528,26 @@ TEST_P(HQDownstreamSessionTest, cancelQPACK) { eventBase_.loopOnce(); } -TEST_P(HQDownstreamSessionTest, DelayedQPACKCanceled) { +void HQDownstreamSessionTest::testDelayedQPACKCanceled(bool eof) { auto req = getGetRequest(); req.getHeaders().add("X-FB-Debug", "rfccffgvtvnenjkbtitkfdufddnvbecu"); - auto id = sendRequest(req); + auto id = sendRequest(req, eof); // This request never gets a handler auto controlStream = encoderWriteBuf_.move(); // receive header block with unsatisfied dep flushRequestsAndLoopN(1); + EXPECT_EQ(socketDriver_->isStreamPaused(id), !eof); // cancel this request socketDriver_->addReadError(id, HTTP3::ErrorCode::HTTP_REQUEST_CANCELLED, std::chrono::milliseconds(0)); flushRequestsAndLoopN(1); + for (size_t i = 0; i < 10 && hqSession_->getNumStreams() != 0; ++i) { + eventBase_.loopOnce(EVLOOP_NONBLOCK); + } + EXPECT_EQ(hqSession_->getNumStreams(), 0); // Now send the dependency encoderWriteBuf_.append(std::move(controlStream)); @@ -2324,6 +2557,40 @@ TEST_P(HQDownstreamSessionTest, DelayedQPACKCanceled) { hqSession_->closeWhenIdle(); } +TEST_P(HQDownstreamSessionTest, DelayedQPACKCanceled) { + testDelayedQPACKCanceled(true); +} + +TEST_P(HQDownstreamSessionTest, DelayedQPACKCanceledWhilePaused) { + testDelayedQPACKCanceled(false); +} + +TEST_P(HQDownstreamSessionTest, QPACKBlockedTimeoutClearsBufferedBody) { + auto req = getPostRequest(4096); + req.getHeaders().add("X-FB-Debug", "rfccffgvtvnenjkbtitkfdufddnvbecu"); + auto id = sendRequest(req, false); + auto& request = getStream(id); + request.codec->generateBody( + request.buf, request.id, makeBuf(4096), HTTPCodec::NoPadding, false); + auto encoderInstructions = encoderWriteBuf_.move(); + ASSERT_NE(encoderInstructions, nullptr); + flushRequestsAndLoopN(1); + ASSERT_TRUE(socketDriver_->isStreamPaused(id)); + + testing::StrictMock handler; + EXPECT_CALL(getMockController(), getTransactionTimeoutHandler(_, _)) + .WillOnce(Return(&handler)); + EXPECT_CALL(handler, _setTransaction(_)).WillOnce(SaveArg<0>(&handler.txn_)); + handler.expectError([&](const HTTPException&) { handler.txn_->sendAbort(); }); + handler.expectDetachTransaction(); + eventBase_.loop(); + EXPECT_EQ(hqSession_->getNumStreams(), 0); + + encoderWriteBuf_.append(std::move(encoderInstructions)); + flushRequestsAndLoop(); + hqSession_->closeWhenIdle(); +} + TEST_P(HQDownstreamSessionTest, DelayedQPACKTimeout) { auto req = getPostRequest(10); req.getHeaders().add("X-FB-Debug", "rfccffgvtvnenjkbtitkfdufddnvbecu"); @@ -2443,6 +2710,33 @@ TEST_P(ControlStreamsStallTest, StalledQpackStream) { hqSession_->closeWhenIdle(); } +TEST_P(HQDownstreamSessionTest, OversizedHeadersFrameHeader) { + hqSession_->setEgressSettings({{SettingsId::MAX_HEADER_LIST_SIZE, 60}}); + auto id = nextStreamId(); + auto res = requests_.emplace(std::piecewise_construct, + std::forward_as_tuple(id), + std::forward_as_tuple(makeCodec(id))); + auto& request = res.first->second; + request.id = request.codec->createStream(); + request.readEOF = false; + hq::writeFrameHeader(request.buf, hq::FrameType::HEADERS, 61); + + testing::StrictMock errHandler; + EXPECT_CALL(getMockController(), getParseErrorHandler(_, _, _)) + .WillOnce(Return(&errHandler)); + EXPECT_CALL(errHandler, _setTransaction(testing::_)) + .WillOnce(testing::SaveArg<0>(&errHandler.txn_)); + errHandler.expectError([&errHandler](const HTTPException& ex) { + EXPECT_EQ(ex.getHttp3ErrorCode(), + (uint32_t)HTTP3::ErrorCode::HTTP_QPACK_DECOMPRESSION_FAILED); + errHandler.sendReplyWithBody(400, 100); + }); + errHandler.expectDetachTransaction(); + flushRequestsAndLoop(); + EXPECT_GT(socketDriver_->streams_[id].writeBuf.chainLength(), 30); + hqSession_->closeWhenIdle(); +} + TEST_P(HQDownstreamSessionTest, QPACKHeadersTooLarge) { hqSession_->setEgressSettings({{SettingsId::MAX_HEADER_LIST_SIZE, 60}}); auto req = getGetRequest(); diff --git a/proxygen/lib/http/session/test/HQDownstreamSessionTest.h b/proxygen/lib/http/session/test/HQDownstreamSessionTest.h index 21e97f774e..5459ff6717 100644 --- a/proxygen/lib/http/session/test/HQDownstreamSessionTest.h +++ b/proxygen/lib/http/session/test/HQDownstreamSessionTest.h @@ -131,6 +131,10 @@ class HQDownstreamSessionTest : public HQSessionTest { void SetUpBase(); void SetUpOnTransportReady(); + void testQPACKBlockedReads(size_t initialBodyBytes); + void testQPACKBlockedReadsWithSessionLimit(bool unblockQPACKFirst); + void testDelayedQPACKCanceled(bool eof); + template std::unique_ptr> addSimpleStrictHandlerBase(); diff --git a/proxygen/lib/http/session/test/HQSessionTestCommon.h b/proxygen/lib/http/session/test/HQSessionTestCommon.h index 760a59ecbf..1b33d60396 100644 --- a/proxygen/lib/http/session/test/HQSessionTestCommon.h +++ b/proxygen/lib/http/session/test/HQSessionTestCommon.h @@ -71,6 +71,13 @@ void createControlStream(quic::MockQuicSocketDriver* socketDriver, quic::StreamId id, proxygen::hq::UnidirectionalStreamType streamType); +class HQDownstreamSessionForTest : public proxygen::HQDownstreamSession { + public: + using HQDownstreamSession::HQDownstreamSession; + using HQSession::findStream; + using HTTPSessionBase::ingressLimitExceeded; +}; + class HQSessionTest : public testing::TestWithParam , public quic::MockQuicSocketDriver::LocalAppCallback @@ -103,7 +110,7 @@ class HQSessionTest { if (direction_ == proxygen::TransportDirection::DOWNSTREAM) { - hqSession_ = new proxygen::HQDownstreamSession( + hqSession_ = new HQDownstreamSessionForTest( std::chrono::milliseconds(kTransactionTimeout), &controllerContainer_.mockController, proxygen::mockTransportInfo, diff --git a/proxygen/lib/http/session/test/HQUpstreamSessionTest.cpp b/proxygen/lib/http/session/test/HQUpstreamSessionTest.cpp index 900f67dafd..f0a37be59f 100644 --- a/proxygen/lib/http/session/test/HQUpstreamSessionTest.cpp +++ b/proxygen/lib/http/session/test/HQUpstreamSessionTest.cpp @@ -344,6 +344,40 @@ TEST_P(HQUpstreamSessionTest, GetWithTrailers) { hqSession_->closeWhenIdle(); } +TEST_P(HQUpstreamSessionTest, QPACKBlockedTrailersResumeReads) { + auto handler = openTransaction(); + handler->txn_->sendHeadersWithEOM(getGetRequest()); + handler->expectHeaders(); + handler->expectBody(); + auto resp = makeResponse(200, 100); + auto id = handler->txn_->getID(); + sendResponse(id, *std::get<0>(resp), std::move(std::get<1>(resp)), false); + flushAndLoopN(1); + + auto& stream = streams_.at(id); + HTTPHeaders trailers; + trailers.add("x-trailer", "qpack-blocked-trailer"); + stream.codec->generateTrailers(stream.buf, stream.codecId, trailers); + auto encoderInstructions = encoderWriteBuf_.move(); + ASSERT_NE(encoderInstructions, nullptr); + flushAndLoopN(1); + EXPECT_TRUE(socketDriver_->isStreamPaused(id)); + + stream.readEOF = true; + flushAndLoopN(1); + stream.readEOF = false; + EXPECT_TRUE(socketDriver_->isStreamPaused(id)); + + handler->expectTrailers(); + handler->expectEOM(); + handler->expectDetachTransaction(); + encoderWriteBuf_.append(std::move(encoderInstructions)); + flushAndLoop(); + EXPECT_EQ(socketDriver_->streams_[id].readState, + MockQuicSocketDriver::CLOSED); + hqSession_->closeWhenIdle(); +} + TEST_P(HQUpstreamSessionTest, AbortOnBodyWithBlockedQPACKTrailers) { auto handler = openTransaction(); auto req = getGetRequest(); @@ -371,6 +405,7 @@ TEST_P(HQUpstreamSessionTest, AbortOnBodyWithBlockedQPACKTrailers) { auto control2 = encoderWriteBuf_.move(); encoderWriteBuf_.append(std::move(control1)); flushAndLoopN(1); + stream.readEOF = false; encoderWriteBuf_.append(std::move(control2)); flushAndLoop();