Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -758,6 +758,38 @@ TEST_F(HTTPTransactionWebTransportTest, StopSendingThenAbort) {
eventBase_.loopOnce();
}

TEST_F(HTTPTransactionWebTransportTest, StopSendingFailsPendingRead) {
auto wtImpl = dynamic_cast<WebTransportImpl*>(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<WebTransportImpl*>(wt_);
ASSERT_NE(wtImpl, nullptr);
Expand Down
6 changes: 5 additions & 1 deletion proxygen/lib/http/webtransport/WebTransportImpl.h
Original file line number Diff line number Diff line change
Expand Up @@ -318,7 +318,11 @@ class WebTransportImpl

folly::Expected<folly::Unit, WebTransport::ErrorCode> 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<folly::IOBuf> data, bool eof);
Expand Down
26 changes: 21 additions & 5 deletions proxygen/lib/http/webtransport/test/FakeSharedWebTransport.h
Original file line number Diff line number Diff line change
Expand Up @@ -76,11 +76,6 @@ class FakeStreamHandle
if (!stopSendingErr_) {
stopSendingErr_ = code;
}
auto& ex = WebTransport::StreamWriteHandle::ex_;
if (!ex) {
ex = folly::make_exception_wrapper<WebTransport::Exception>(code);
WebTransport::StreamWriteHandle::cs_.requestCancellation();
}
if (!writeErr_) {
writeErr_.emplace(code);
}
Expand All @@ -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<WebTransport::Exception>(code);
WebTransport::StreamWriteHandle::cs_.requestCancellation();
}
return folly::unit;
}

Expand Down Expand Up @@ -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(
Expand Down
Loading