diff --git a/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp b/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp index 514f981c40..aff61d36b1 100644 --- a/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp +++ b/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp @@ -758,6 +758,38 @@ TEST_F(HTTPTransactionWebTransportTest, StopSendingThenAbort) { eventBase_.loopOnce(); } +TEST_F(HTTPTransactionWebTransportTest, StopSendingFailsPendingRead) { + auto wtImpl = dynamic_cast(wt_); + ASSERT_NE(wtImpl, nullptr); + wtImpl->setUniStreamFlowControl( + /*maxStreamId=*/4, + /*targetConcurrentStreams=*/4); + + WebTransport::StreamReadHandle* readHandle{nullptr}; + EXPECT_CALL(handler_, onWebTransportUniStream(_, _)) + .WillOnce(SaveArg<1>(&readHandle)); + + txn_->onWebTransportUniStream(2); + ASSERT_NE(readHandle, nullptr); + + auto fut = + readHandle->readStreamData() + .via(&eventBase_) + .thenTry([](auto streamData) { + readCallback(std::move(streamData), true, 0, false, WT_APP_ERROR_2); + }); + EXPECT_FALSE(fut.isReady()); + + EXPECT_CALL(transport_, + stopReadingWebTransportIngress(2, makeOpt(WT_APP_ERROR_2))) + .WillOnce(Return(folly::unit)); + readHandle->stopSending(WT_APP_ERROR_2); + // The transport dropped the read callback, so the peer's RESET_STREAM + // cannot complete the read; stopSending must. + eventBase_.loopOnce(); + EXPECT_TRUE(fut.isReady()); +} + TEST_F(HTTPTransactionWebTransportTest, WriteBufferingBasic) { auto wtImpl = dynamic_cast(wt_); ASSERT_NE(wtImpl, nullptr); diff --git a/proxygen/lib/http/webtransport/WebTransportImpl.h b/proxygen/lib/http/webtransport/WebTransportImpl.h index 3f0746a1be..4eb50c2c75 100644 --- a/proxygen/lib/http/webtransport/WebTransportImpl.h +++ b/proxygen/lib/http/webtransport/WebTransportImpl.h @@ -318,7 +318,11 @@ class WebTransportImpl folly::Expected stopSending( uint32_t error) override { - return impl_.stopReadingWebTransportIngress(id_, error); + auto res = impl_.stopReadingWebTransportIngress(id_, error); + // The read callback is gone, so the solicited RESET_STREAM cannot fail + // a pending read. May destroy this handle. + deliverReadError(WebTransport::Exception(error, "sent stop_sending")); + return res; } FCState dataAvailable(std::unique_ptr data, bool eof); diff --git a/proxygen/lib/http/webtransport/test/FakeSharedWebTransport.h b/proxygen/lib/http/webtransport/test/FakeSharedWebTransport.h index 68e83b569c..a9e5cc74ce 100644 --- a/proxygen/lib/http/webtransport/test/FakeSharedWebTransport.h +++ b/proxygen/lib/http/webtransport/test/FakeSharedWebTransport.h @@ -76,11 +76,6 @@ class FakeStreamHandle if (!stopSendingErr_) { stopSendingErr_ = code; } - auto& ex = WebTransport::StreamWriteHandle::ex_; - if (!ex) { - ex = folly::make_exception_wrapper(code); - WebTransport::StreamWriteHandle::cs_.requestCancellation(); - } if (!writeErr_) { writeErr_.emplace(code); } @@ -90,6 +85,12 @@ class FakeStreamHandle auto cb = std::move(onWriteTerminal_); cb(); } + // Last: a synchronous peer reaction can destroy this handle. + auto& ex = WebTransport::StreamWriteHandle::ex_; + if (!ex) { + ex = folly::make_exception_wrapper(code); + WebTransport::StreamWriteHandle::cs_.requestCancellation(); + } return folly::unit; } @@ -239,6 +240,21 @@ class FakeStreamHandle WebTransport::StreamWriteHandle::cs_.requestCancellation(); return folly::unit; } + + // Simulate a peer RESET_STREAM on this read half. The writer's state stays as + // if it had not reset. + void deliverPeerReset(uint32_t err) { + if (promise_) { + promise_->setException(WebTransport::Exception(err)); + promise_.reset(); + } + writeErr_ = err; + if (!fin_ && onWriteTerminal_) { + auto cb = std::move(onWriteTerminal_); + cb(); + } + } + GenericApiRet setPriority(quic::PriorityQueue::Priority priority) override { quic::HTTPPriorityQueue::Priority httpPri(priority); pri.emplace(std::forward_as_tuple(