Skip to content

Commit 89beea0

Browse files
afrindmeta-codesync[bot]
afrind
authored and
meta-codesync[bot]
committed
Fix spurious REQUEST_UPDATE delivery after session close
Summary: handleRequestUpdate coroutines launched via co_withExecutor(..).start() could run after the session had already closed, crashing in setRequestSession() (shared_from_this() on a destroyed session). This is the same class of bug fixed in D95335320 for handleSubscribe, handlePublish, handleFetch, and handleTrackStatus — but for the REQUEST_UPDATE path in TrackPublisherImpl::onRequestUpdate, FetchPublisherImpl (via handleFetchRequestUpdate), and MoQRelaySession's publish/subscribe namespace request update handlers. Fix: wrap each co_withExecutor launch with co_withCancellation and add co_await co_safe_point at the top of the coroutine body, so the coroutine bails out immediately if the cancellation token has been triggered by cleanup(). Reviewed By: sandarsh Differential Revision: D95508365 fbshipit-source-id: 9b23f42a521528f4ed4a1fb0d7c80c06ea21d7c0
1 parent b5bb96f commit 89beea0

3 files changed

Lines changed: 139 additions & 73 deletions

File tree

‎moxygen/MoQRelaySession.cpp‎

Lines changed: 73 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -358,37 +358,42 @@ void MoQRelaySession::handlePublishNamespaceRequestUpdate(
358358
// Handle asynchronously with shared ownership to prevent use-after-free
359359
co_withExecutor(
360360
getExecutor(),
361-
folly::coro::co_invoke(
362-
[this,
363-
announceHandle = std::move(announceHandle),
364-
update = std::move(requestUpdate),
365-
existingRequestID,
366-
updateRequestID]() mutable -> folly::coro::Task<void> {
367-
// Call the handle's requestUpdate
368-
auto updateResult = co_await co_awaitTry(co_withCancellation(
369-
cancellationSource_.getToken(),
370-
announceHandle->requestUpdate(std::move(update))));
371-
372-
// Only send responses for v15+
373-
if (getDraftMajorVersion(*getNegotiatedVersion()) >= 15) {
374-
if (updateResult.hasException()) {
375-
XLOG(ERR) << "Exception in requestUpdate ex="
376-
<< updateResult.exception().what();
377-
requestUpdateError(
378-
RequestError{
379-
updateRequestID,
380-
RequestErrorCode::INTERNAL_ERROR,
381-
"Exception in requestUpdate"},
382-
existingRequestID);
383-
} else if (updateResult->hasError()) {
384-
auto updateErr = updateResult->error();
385-
requestUpdateError(updateErr, existingRequestID);
386-
} else {
387-
RequestOk requestOk{.requestID = updateRequestID};
388-
requestUpdateOk(requestOk);
389-
}
390-
}
391-
}))
361+
co_withCancellation(
362+
cancellationSource_.getToken(),
363+
folly::coro::co_invoke(
364+
[this,
365+
announceHandle = std::move(announceHandle),
366+
update = std::move(requestUpdate),
367+
existingRequestID,
368+
updateRequestID]() mutable -> folly::coro::Task<void> {
369+
co_await folly::coro::co_safe_point;
370+
// Call the handle's requestUpdate
371+
auto updateResult = co_await co_awaitTry(co_withCancellation(
372+
cancellationSource_.getToken(),
373+
announceHandle->requestUpdate(std::move(update))));
374+
375+
// Only send responses for v15+
376+
if (getDraftMajorVersion(*getNegotiatedVersion()) >= 15) {
377+
if (updateResult.hasException()) {
378+
XLOG(ERR) << "Exception in requestUpdate ex="
379+
<< updateResult.exception().what();
380+
requestUpdateError(
381+
RequestError{
382+
updateRequestID,
383+
RequestErrorCode::INTERNAL_ERROR,
384+
"Exception in requestUpdate"},
385+
existingRequestID);
386+
} else if (updateResult->hasError()) {
387+
auto updateErr = updateResult->error();
388+
requestUpdateError(updateErr, existingRequestID);
389+
} else {
390+
RequestOk requestOk{
391+
.requestID = updateRequestID,
392+
.requestSpecificParams = {}};
393+
requestUpdateOk(requestOk);
394+
}
395+
}
396+
})))
392397
.start();
393398
}
394399

@@ -405,37 +410,43 @@ void MoQRelaySession::handleSubscribeNamespaceRequestUpdate(
405410
// Handle asynchronously with shared ownership to prevent use-after-free
406411
co_withExecutor(
407412
getExecutor(),
408-
folly::coro::co_invoke(
409-
[this,
410-
subscribeNamespaceHandle = std::move(subscribeNamespaceHandle),
411-
update = std::move(requestUpdate),
412-
existingRequestID,
413-
updateRequestID]() mutable -> folly::coro::Task<void> {
414-
// Call the handle's requestUpdate
415-
auto updateResult = co_await co_awaitTry(co_withCancellation(
416-
cancellationSource_.getToken(),
417-
subscribeNamespaceHandle->requestUpdate(std::move(update))));
418-
419-
// Only send responses for v15+
420-
if (getDraftMajorVersion(*getNegotiatedVersion()) >= 15) {
421-
if (updateResult.hasException()) {
422-
XLOG(ERR) << "Exception in requestUpdate ex="
423-
<< updateResult.exception().what();
424-
requestUpdateError(
425-
RequestError{
426-
updateRequestID,
427-
RequestErrorCode::INTERNAL_ERROR,
428-
"Exception in requestUpdate"},
429-
existingRequestID);
430-
} else if (updateResult->hasError()) {
431-
auto updateErr = updateResult->error();
432-
requestUpdateError(updateErr, existingRequestID);
433-
} else {
434-
RequestOk requestOk{.requestID = updateRequestID};
435-
requestUpdateOk(requestOk);
436-
}
437-
}
438-
}))
413+
co_withCancellation(
414+
cancellationSource_.getToken(),
415+
folly::coro::co_invoke(
416+
[this,
417+
subscribeNamespaceHandle = std::move(subscribeNamespaceHandle),
418+
update = std::move(requestUpdate),
419+
existingRequestID,
420+
updateRequestID]() mutable -> folly::coro::Task<void> {
421+
co_await folly::coro::co_safe_point;
422+
// Call the handle's requestUpdate
423+
auto updateResult = co_await co_awaitTry(co_withCancellation(
424+
cancellationSource_.getToken(),
425+
subscribeNamespaceHandle->requestUpdate(
426+
std::move(update))));
427+
428+
// Only send responses for v15+
429+
if (getDraftMajorVersion(*getNegotiatedVersion()) >= 15) {
430+
if (updateResult.hasException()) {
431+
XLOG(ERR) << "Exception in requestUpdate ex="
432+
<< updateResult.exception().what();
433+
requestUpdateError(
434+
RequestError{
435+
updateRequestID,
436+
RequestErrorCode::INTERNAL_ERROR,
437+
"Exception in requestUpdate"},
438+
existingRequestID);
439+
} else if (updateResult->hasError()) {
440+
auto updateErr = updateResult->error();
441+
requestUpdateError(updateErr, existingRequestID);
442+
} else {
443+
RequestOk requestOk{
444+
.requestID = updateRequestID,
445+
.requestSpecificParams = {}};
446+
requestUpdateOk(requestOk);
447+
}
448+
}
449+
})))
439450
.start();
440451
}
441452

‎moxygen/MoQSession.cpp‎

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1116,16 +1116,19 @@ class MoQSession::TrackPublisherImpl : public MoQSession::PublisherImpl,
11161116
// if session closes before completion
11171117
co_withExecutor(
11181118
session_->getExecutor(),
1119-
folly::coro::co_invoke(
1120-
[trackPubImpl = std::move(trackPubImpl),
1121-
update = std::move(
1122-
requestUpdate)]() mutable -> folly::coro::Task<void> {
1123-
co_await trackPubImpl->handleRequestUpdate(std::move(update));
1124-
}))
1119+
co_withCancellation(
1120+
session_->cancellationSource_.getToken(),
1121+
folly::coro::co_invoke(
1122+
[trackPubImpl = std::move(trackPubImpl),
1123+
update = std::move(
1124+
requestUpdate)]() mutable -> folly::coro::Task<void> {
1125+
co_await trackPubImpl->handleRequestUpdate(std::move(update));
1126+
})))
11251127
.start();
11261128
}
11271129

11281130
folly::coro::Task<void> handleRequestUpdate(RequestUpdate requestUpdate) {
1131+
co_await folly::coro::co_safe_point;
11291132
folly::RequestContextScopeGuard guard;
11301133
session_->setRequestSession();
11311134

@@ -3493,11 +3496,14 @@ void MoQSession::handleFetchRequestUpdate(
34933496
// Simple passthrough - just deliver to application and relay response
34943497
co_withExecutor(
34953498
getExecutor(),
3496-
folly::coro::co_invoke(
3497-
[fetchPublisher, update = std::move(requestUpdate)]() mutable
3498-
-> folly::coro::Task<void> {
3499-
co_await fetchPublisher->onRequestUpdate(std::move(update));
3500-
}))
3499+
co_withCancellation(
3500+
cancellationSource_.getToken(),
3501+
folly::coro::co_invoke(
3502+
[fetchPublisher = fetchPublisher,
3503+
update = requestUpdate]() mutable -> folly::coro::Task<void> {
3504+
co_await folly::coro::co_safe_point;
3505+
co_await fetchPublisher->onRequestUpdate(std::move(update));
3506+
})))
35013507
.start();
35023508
}
35033509

‎moxygen/test/MoQSessionRequestUpdateTests.cpp‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -339,4 +339,53 @@ CO_TEST_P_X(MoQSessionTest, FetchRequestUpdateNotSupported) {
339339
clientSession_->close(SessionCloseErrorCode::NO_ERROR);
340340
}
341341

342+
// Regression test: when a REQUEST_UPDATE is queued on the executor but the
343+
// session closes before it runs, the update should not be delivered to the
344+
// application handler. cleanup() calls unsubscribe() on the handle; the
345+
// subsequent requestUpdate should be suppressed.
346+
CO_TEST_P_X(MoQSessionTest, RequestUpdateAfterClose) {
347+
co_await setupMoQSession();
348+
std::shared_ptr<MockSubscriptionHandle> mockSubscriptionHandle = nullptr;
349+
350+
// Cleanup will deliver a publishDone to the client's subscribeCallback_
351+
EXPECT_CALL(*subscribeCallback_, publishDone(_))
352+
.WillOnce(testing::Return(folly::unit));
353+
expectSubscribe(
354+
[&mockSubscriptionHandle](auto sub, auto /*pub*/) -> TaskSubscribeResult {
355+
mockSubscriptionHandle =
356+
makeSubscribeOkResult(sub, AbsoluteLocation{0, 0});
357+
co_return mockSubscriptionHandle;
358+
});
359+
360+
auto subscribeRequest = getSubscribe(kTestTrackName);
361+
auto res =
362+
co_await clientSession_->subscribe(subscribeRequest, subscribeCallback_);
363+
EXPECT_FALSE(res.hasError());
364+
365+
auto* cb =
366+
static_cast<MoQControlCodec::ControlCallback*>(serverSession_.get());
367+
368+
// Simulate a REQUEST_UPDATE arriving at the server for the existing
369+
// subscribe (requestID 0). This queues handleRequestUpdate on the
370+
// executor but it won't run until we yield.
371+
RequestUpdate update;
372+
update.existingRequestID = subscribeRequest.requestID;
373+
update.requestID = RequestID(getRequestIDMultiplier());
374+
update.priority = kDefaultPriority + 1;
375+
update.forward = true;
376+
cb->onRequestUpdate(std::move(update));
377+
378+
// requestUpdateCalled must NOT be invoked — the session is about to close
379+
EXPECT_CALL(*mockSubscriptionHandle, requestUpdateCalled).Times(0);
380+
381+
// Close the session before handleRequestUpdate runs. Use a PUBLISH with
382+
// wrong requestID parity to trigger closeSessionIfRequestIDInvalid.
383+
cb->onPublish(PublishRequest{.requestID = RequestID(1), .fullTrackName = {}});
384+
385+
// Yield to let queued coroutines run
386+
co_await folly::coro::co_reschedule_on_current_executor;
387+
co_await folly::coro::co_reschedule_on_current_executor;
388+
co_await folly::coro::co_reschedule_on_current_executor;
389+
}
390+
342391
} // namespace

0 commit comments

Comments
 (0)