Skip to content

Commit 7aa8849

Browse files
afrindakash-a-nclaude
committed
add onSubscriptionBegin/onSubscriptionEnd gauge stats; fix onPublishDone semantics
- onPublishDone fires only when a PUBLISH_DONE wire frame is received/sent, not for session-close synthesized cases - Add onSubscriptionBegin/onSubscriptionEnd to MoQStatsCallback (common base) for an active-subscription gauge that can never go negative; fires on both SUBSCRIBE and PUBLISH paths (publisher and subscriber sides) - onSubscriptionEnd fires at every subscription termination path: subscriber: deliverPublishDoneAndRemove, unsubscribe publisher: sendPublishDone, onUnsubscribe, cleanup - Switch stats test mocks to NiceMock to suppress uninteresting-call warnings Co-Authored-By: Akash A N <akashn@akamai.com> Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent b770f30 commit 7aa8849

5 files changed

Lines changed: 80 additions & 4 deletions

File tree

‎moxygen/MoQSession.cpp‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1968,6 +1968,8 @@ class MoQSession::SubscribeTrackReceiveState
19681968
<< "deliverPublishDoneAndRemove: Delivering PUBLISH_DONE to app; statusCode="
19691969
<< folly::to_underlying(pendingPublishDone_->statusCode)
19701970
<< " alias=" << alias_ << " requestID=" << requestID_;
1971+
MOQ_SUBSCRIBER_STATS(
1972+
session_->subscriberStatsCallback_, onSubscriptionEnd);
19711973
auto token = cancelSource_.getToken();
19721974
auto cb = std::exchange(callback_, nullptr);
19731975
cb->publishDone(std::move(*pendingPublishDone_));
@@ -2205,6 +2207,7 @@ void MoQSession::cleanup() {
22052207
auto requestID = it->first;
22062208
auto pubTrack = std::move(it->second);
22072209
pubTracks_.erase(it);
2210+
MOQ_PUBLISHER_STATS(publisherStatsCallback_, onSubscriptionEnd);
22082211
pubTrack->terminatePublish(
22092212
PublishDone(
22102213
{requestID,
@@ -3668,6 +3671,7 @@ void MoQSession::onUnsubscribe(Unsubscribe unsubscribe) {
36683671
} else {
36693672
trackPublisher->unsubscribe();
36703673
if (pubTracks_.erase(unsubscribe.requestID)) {
3674+
MOQ_PUBLISHER_STATS(publisherStatsCallback_, onSubscriptionEnd);
36713675
retireRequestID(/*signalWriteLoop=*/true);
36723676
} // else, the caller invoked publishDone, which isn't needed but fine
36733677
}
@@ -3703,6 +3707,7 @@ void MoQSession::onPublishOk(PublishOk publishOk) {
37033707
std::static_pointer_cast<TrackPublisherImpl>(trackIt->second);
37043708
trackPublisher->onPublishOk(publishOk);
37053709
pendingPublishTracks_.erase(trackIt->second->fullTrackName());
3710+
MOQ_PUBLISHER_STATS(publisherStatsCallback_, onSubscriptionBegin);
37063711
}
37073712

37083713
publishPtr->setValue(std::move(publishOk));
@@ -4763,6 +4768,7 @@ Subscriber::PublishResult MoQSession::publish(
47634768
void MoQSession::publishOk(const PublishOk& pubOk, ReplyContext& replyContext) {
47644769
XLOG(DBG1) << __func__ << " reqID=" << pubOk.requestID << " sess=" << this;
47654770
MOQ_SUBSCRIBER_STATS(subscriberStatsCallback_, onPublishOk);
4771+
MOQ_SUBSCRIBER_STATS(subscriberStatsCallback_, onSubscriptionBegin);
47664772

47674773
if (logger_) {
47684774
logger_->logPublishOk(pubOk, ControlMessageType::CREATED);
@@ -4885,6 +4891,7 @@ folly::coro::Task<Publisher::SubscribeResult> MoQSession::subscribe(
48854891
co_return folly::makeUnexpected(subscribeResult.error());
48864892
} else {
48874893
MOQ_SUBSCRIBER_STATS(subscriberStatsCallback_, onSubscribeSuccess);
4894+
MOQ_SUBSCRIBER_STATS(subscriberStatsCallback_, onSubscriptionBegin);
48884895
co_return std::make_shared<ReceiverSubscriptionHandle>(
48894896
std::move(subscribeResult.value()), trackAlias, shared_from_this());
48904897
}
@@ -4893,6 +4900,7 @@ folly::coro::Task<Publisher::SubscribeResult> MoQSession::subscribe(
48934900
void MoQSession::sendSubscribeOk(const SubscribeOk& subOk, ReplyContext& ctx) {
48944901
XLOG(DBG1) << __func__ << " sess=" << this;
48954902
MOQ_PUBLISHER_STATS(publisherStatsCallback_, onSubscribeSuccess);
4903+
MOQ_PUBLISHER_STATS(publisherStatsCallback_, onSubscriptionBegin);
48964904
auto res = moqFrameWriter_.writeSubscribeOk(ctx.writeBuf(), subOk);
48974905
if (!res) {
48984906
XLOG(ERR) << "writeSubscribeOk failed sess=" << this;
@@ -4951,6 +4959,9 @@ void MoQSession::unsubscribe(const Unsubscribe& unsubscribe) {
49514959
<< " sess=" << this;
49524960
// cancel() should send STOP_SENDING on any open streams for this
49534961
// subscription
4962+
if (trackIt->second->getSubscribeCallback()) {
4963+
MOQ_SUBSCRIBER_STATS(subscriberStatsCallback_, onSubscriptionEnd);
4964+
}
49544965
trackIt->second->cancel();
49554966
subTracks_.erase(trackIt);
49564967
reqIdToTrackAlias_.erase(trackAliasIt);
@@ -4979,6 +4990,7 @@ void MoQSession::sendPublishDone(const PublishDone& pubDone) {
49794990
<< " sess=" << this;
49804991
return;
49814992
}
4993+
MOQ_PUBLISHER_STATS(publisherStatsCallback_, onSubscriptionEnd);
49824994
auto* ctx = it->second->replyContext();
49834995
SCOPE_EXIT {
49844996
pubTracks_.erase(it);

‎moxygen/stats/MoQStats.h‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,25 @@ class MoQStatsCallback {
125125
* Subscriber: The publisher closed a subscription stream to us
126126
*/
127127
virtual void onSubscriptionStreamClosed() = 0;
128+
129+
/*
130+
* Publisher: A subscription became active (sent SUBSCRIBE_OK, or had a
131+
* PUBLISH accepted via PUBLISH_OK)
132+
* Subscriber: A subscription became active (received SUBSCRIBE_OK, or sent
133+
* PUBLISH_OK accepting a PUBLISH)
134+
* Pairs with onSubscriptionEnd to form a gauge of active subscriptions.
135+
*/
136+
virtual void onSubscriptionBegin() = 0;
137+
138+
/*
139+
* Publisher: An active subscription ended (sent PUBLISH_DONE, received
140+
* UNSUBSCRIBE, or session closed while subscription was active)
141+
* Subscriber: An active subscription ended (PUBLISH_DONE delivered to app,
142+
* unsubscribe called, or session closed while subscription was active)
143+
* Pairs with onSubscriptionBegin; this callback fires exactly once per
144+
* onSubscriptionBegin, so the difference is a non-negative gauge.
145+
*/
146+
virtual void onSubscriptionEnd() = 0;
128147
};
129148

130149
class MoQPublisherStatsCallback : public MoQStatsCallback {

‎moxygen/test/MoQSessionSubscribeTests.cpp‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -806,6 +806,10 @@ CO_TEST_P_X(MoQSessionTest, UnsubscribeWithinPublishDone) {
806806
auto subscribeHandler = res.value();
807807

808808
EXPECT_CALL(*clientSubscriberStatsCallback_, onUnsubscribe());
809+
EXPECT_CALL(
810+
*clientSubscriberStatsCallback_,
811+
onPublishDone(PublishDoneStatusCode::TRACK_ENDED));
812+
EXPECT_CALL(*clientSubscriberStatsCallback_, onSubscriptionEnd());
809813

810814
EXPECT_CALL(*subscribeCallback_, publishDone(_))
811815
.WillOnce(testing::Invoke([&](const auto&) {
@@ -1020,6 +1024,8 @@ CO_TEST_P_X(MoQSessionTest, UnsubscribeFromWithinPublishDoneHandler) {
10201024
// When publishDone is delivered, immediately call unsubscribe() from
10211025
// handler
10221026
folly::coro::Baton publishDoneInvoked;
1027+
// onPublishDone fires before this point (wire handler); NiceMock handles it.
1028+
EXPECT_CALL(*clientSubscriberStatsCallback_, onSubscriptionEnd());
10231029
EXPECT_CALL(*subscribeCallback_, publishDone(_))
10241030
.WillOnce(testing::Invoke([&](const auto&) {
10251031
subscribeHandle->unsubscribe();
@@ -1107,3 +1113,26 @@ CO_TEST_P_X(MoQSessionTest, BeginSubgroupCallbackError) {
11071113
co_await publishDone_;
11081114
clientSession_->close(SessionCloseErrorCode::NO_ERROR);
11091115
}
1116+
1117+
// Regression: session close fires onSubscriptionEnd for abandoned publishers;
1118+
// onPublishDone must NOT fire — no PUBLISH_DONE wire frame was ever received.
1119+
CO_TEST_P_X(MoQSessionTest, SubscriptionEndStatFiredOnAbandonedPublisher) {
1120+
co_await setupMoQSession();
1121+
expectSubscribe([](auto sub, auto) -> TaskSubscribeResult {
1122+
// Publisher accepts the subscription but intentionally never calls
1123+
// publishDone, simulating an abandoned / improperly-closed publisher.
1124+
co_return makeSubscribeOkResult(sub);
1125+
});
1126+
1127+
auto res = co_await clientSession_->subscribe(
1128+
getSubscribe(kTestTrackName), subscribeCallback_);
1129+
EXPECT_FALSE(res.hasError());
1130+
1131+
EXPECT_CALL(*clientSubscriberStatsCallback_, onPublishDone(_)).Times(0);
1132+
EXPECT_CALL(*clientSubscriberStatsCallback_, onSubscriptionEnd());
1133+
EXPECT_CALL(*subscribeCallback_, publishDone(_))
1134+
.WillOnce(testing::Return(folly::unit));
1135+
1136+
clientSession_->close(SessionCloseErrorCode::NO_ERROR);
1137+
serverSession_->close(SessionCloseErrorCode::NO_ERROR);
1138+
}

‎moxygen/test/MoQSessionTestCommon.cpp‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -194,16 +194,20 @@ void MoQSessionTest::SetUp() {
194194
folly::Expected<folly::Unit, MoQPublishError>(folly::unit)));
195195
EXPECT_CALL(*subscribeCallback_, setTrackAlias(_)).Times(testing::AtLeast(0));
196196

197-
clientSubscriberStatsCallback_ = std::make_shared<MockSubscriberStats>();
197+
clientSubscriberStatsCallback_ =
198+
std::make_shared<testing::NiceMock<MockSubscriberStats>>();
198199
clientSession_->setSubscriberStatsCallback(clientSubscriberStatsCallback_);
199200

200-
clientPublisherStatsCallback_ = std::make_shared<MockPublisherStats>();
201+
clientPublisherStatsCallback_ =
202+
std::make_shared<testing::NiceMock<MockPublisherStats>>();
201203
clientSession_->setPublisherStatsCallback(clientPublisherStatsCallback_);
202204

203-
serverSubscriberStatsCallback_ = std::make_shared<MockSubscriberStats>();
205+
serverSubscriberStatsCallback_ =
206+
std::make_shared<testing::NiceMock<MockSubscriberStats>>();
204207
serverSession_->setSubscriberStatsCallback(serverSubscriberStatsCallback_);
205208

206-
serverPublisherStatsCallback_ = std::make_shared<MockPublisherStats>();
209+
serverPublisherStatsCallback_ =
210+
std::make_shared<testing::NiceMock<MockPublisherStats>>();
207211
serverSession_->setPublisherStatsCallback(serverPublisherStatsCallback_);
208212

209213
// For Draft15+, initialize version via ALPN since it's required
@@ -436,7 +440,11 @@ void MoQSessionTest::expectPublishDone(MoQControlCodec::Direction recipient) {
436440
EXPECT_CALL(
437441
*getPublisherStatsCallback(oppositeDirection(recipient)),
438442
onPublishDone(_));
443+
EXPECT_CALL(
444+
*getPublisherStatsCallback(oppositeDirection(recipient)),
445+
onSubscriptionEnd());
439446
EXPECT_CALL(*getSubscriberStatsCallback(recipient), onPublishDone(_));
447+
EXPECT_CALL(*getSubscriberStatsCallback(recipient), onSubscriptionEnd());
440448
EXPECT_CALL(*subscribeCallback_, publishDone(_)).WillOnce([&] {
441449
publishDone_.post();
442450
return folly::unit;

‎moxygen/test/Mocks.h‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -422,6 +422,10 @@ class MockPublisherStats : public MoQPublisherStatsCallback {
422422

423423
MOCK_METHOD(void, onSubscriptionStreamClosed, (), (override));
424424

425+
MOCK_METHOD(void, onSubscriptionBegin, (), (override));
426+
427+
MOCK_METHOD(void, onSubscriptionEnd, (), (override));
428+
425429
MOCK_METHOD(void, recordPublishNamespaceLatency, (uint64_t), (override));
426430

427431
MOCK_METHOD(void, recordPublishLatency, (uint64_t), (override));
@@ -477,6 +481,10 @@ class MockSubscriberStats : public MoQSubscriberStatsCallback {
477481

478482
MOCK_METHOD(void, onSubscriptionStreamClosed, (), (override));
479483

484+
MOCK_METHOD(void, onSubscriptionBegin, (), (override));
485+
486+
MOCK_METHOD(void, onSubscriptionEnd, (), (override));
487+
480488
MOCK_METHOD(void, recordSubscribeLatency, (uint64_t), (override));
481489

482490
MOCK_METHOD(void, recordFetchLatency, (uint64_t), (override));

0 commit comments

Comments
 (0)