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();