Skip to content

Commit 1682d86

Browse files
afrindmeta-codesync[bot]
afrind
authored and
meta-codesync[bot]
committed
fix: don't re-apply range.start check to existing subgroup consumers (#156)
Summary: Fixes a race condition in MoQForwarder::SubgroupForwarder::forEachSubscriberSubgroup where the full checkRange was applied to already-open subgroup consumers. This caused objects to be dropped mid-stream when range.start advanced (e.g. via a LargestObject in onPublishOk) after a subgroup was already opened: the check would incorrectly conclude the subscriber was out of range and skip delivery. The fix separates the two cases: for subscribers with an existing subgroup consumer, only checkPastEnd is applied (which is still correct — we should stop delivering once we're past range.end); the full checkRange gating is reserved for new subgroup creation. A new checkPastEnd helper is extracted from checkRange to make this split clean. Also removes countReceivedObject (tracking state that was unused), drops the now-unnecessary FullTrackName parameter from several MoQCache internal methods, and moves the range-race regression test from MoQCacheTests into MoQForwarderTest where it belongs. Pull Request resolved: #156 Reviewed By: sandarsh Differential Revision: D103879348 Pulled By: afrind fbshipit-source-id: c2931180939aefe5c913d33b325fde3fed49bddb
1 parent baf47f7 commit 1682d86

3 files changed

Lines changed: 159 additions & 58 deletions

File tree

‎moxygen/relay/MoQForwarder.cpp‎

Lines changed: 74 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -59,60 +59,71 @@ MoQForwarder::SubgroupForwarder::forEachSubscriberSubgroup(
5959
return folly::makeUnexpected(
6060
MoQPublishError(MoQPublishError::CANCELLED, "Forwarder detached"));
6161
}
62+
if (!forwarder_->largest_) {
63+
return folly::makeUnexpected(
64+
MoQPublishError(MoQPublishError::CANCELLED, "No subscribers"));
65+
}
6266
bool anyForwarded = false;
6367
forwarder_->forEachSubscriber([&](const std::shared_ptr<Subscriber>& sub) {
64-
if (forwarder_->largest_ && forwarder_->checkRange(*sub)) {
65-
auto subgroupConsumerIt = sub->subgroups.find(identifier_);
66-
if (subgroupConsumerIt != sub->subgroups.end()) {
67-
// Entry exists - check if it's tombstoned (nullptr)
68-
if (!subgroupConsumerIt->second) {
69-
// Tombstoned - skip this subscriber for this subgroup
70-
XLOG(DBG2) << "Skipping tombstoned subgroup for sub=" << sub.get();
71-
return;
72-
}
73-
// Has valid consumer - continue with existing logic
74-
if (!sub->checkShouldForward()) {
75-
// If we're attempting to send anything on an existing subgroup when
76-
// forward == false, then we reset the stream, so that we don't end
77-
// up with "holes" in the subgroup. If, at some point in the future,
78-
// we set forward = true, then we'll create a new stream for the
79-
// subgroup.
80-
subgroupConsumerIt->second->reset(
81-
ResetStreamErrorCode::INTERNAL_ERROR);
82-
closeSubgroupForSubscriber(
83-
sub, "SubgroupForwarder::forEachSubscriberSubgroup");
84-
} else {
85-
anyForwarded = true;
86-
fn(sub, subgroupConsumerIt->second);
87-
}
68+
auto subgroupConsumerIt = sub->subgroups.find(identifier_);
69+
if (subgroupConsumerIt != sub->subgroups.end()) {
70+
// For an existing consumer, only past-end retires it. If range.start
71+
// advanced within this subgroup (e.g. LargestObject onPublishOk), we
72+
// still deliver all objects: within a subgroup, later objects depend on
73+
// earlier ones so delivering more than the subscriber asked is correct.
74+
// TODO: if range.start advanced past this subgroup's group entirely
75+
// (e.g. via SubscribeUpdate), we should reset the open subgroup rather
76+
// than continuing to deliver to it.
77+
if (forwarder_->checkPastEnd(*sub)) {
78+
return;
79+
}
80+
if (!subgroupConsumerIt->second) {
81+
// Tombstoned - skip this subscriber for this subgroup
82+
XLOG(DBG2) << "Skipping tombstoned subgroup for sub=" << sub.get();
83+
return;
84+
}
85+
if (!sub->checkShouldForward()) {
86+
// If we're attempting to send anything on an existing subgroup when
87+
// forward == false, then we reset the stream, so that we don't end
88+
// up with "holes" in the subgroup. If, at some point in the future,
89+
// we set forward = true, then we'll create a new stream for the
90+
// subgroup.
91+
subgroupConsumerIt->second->reset(ResetStreamErrorCode::INTERNAL_ERROR);
92+
closeSubgroupForSubscriber(
93+
sub, "SubgroupForwarder::forEachSubscriberSubgroup");
8894
} else {
89-
// Entry doesn't exist - late joiner logic (create new subgroup if
90-
// makeNew)
91-
if (!sub->checkShouldForward()) {
92-
// If shouldForward == false, we shouldn't be creating any
93-
// subgroups.
94-
return;
95-
}
96-
if (!makeNew) {
97-
XLOG(DBG2) << "skipping creating subgroup for sub=" << sub.get();
98-
return;
99-
}
100-
XCHECK(sub->trackConsumer);
101-
XLOG(DBG2) << "Making new subgroup for consumer=" << sub->trackConsumer
102-
<< " " << callsite;
103-
auto res = sub->trackConsumer->beginSubgroup(
104-
identifier_.group,
105-
identifier_.subgroup,
106-
priority_,
107-
containsLastInGroup_);
108-
if (res.hasError()) {
109-
forwarder_->removeSubscriberOnError(*sub, res.error(), callsite);
110-
} else {
111-
auto emplaceRes = sub->subgroups.emplace(identifier_, res.value());
112-
subgroupConsumerIt = emplaceRes.first;
113-
anyForwarded = true;
114-
fn(sub, subgroupConsumerIt->second);
115-
}
95+
anyForwarded = true;
96+
fn(sub, subgroupConsumerIt->second);
97+
}
98+
} else {
99+
// No consumer yet: full range check gates new subgroup creation.
100+
if (!forwarder_->checkRange(*sub)) {
101+
return;
102+
}
103+
if (!sub->checkShouldForward()) {
104+
// If shouldForward == false, we shouldn't be creating any
105+
// subgroups.
106+
return;
107+
}
108+
if (!makeNew) {
109+
XLOG(DBG2) << "skipping creating subgroup for sub=" << sub.get();
110+
return;
111+
}
112+
XCHECK(sub->trackConsumer);
113+
XLOG(DBG2) << "Making new subgroup for consumer=" << sub->trackConsumer
114+
<< " " << callsite;
115+
auto res = sub->trackConsumer->beginSubgroup(
116+
identifier_.group,
117+
identifier_.subgroup,
118+
priority_,
119+
containsLastInGroup_);
120+
if (res.hasError()) {
121+
forwarder_->removeSubscriberOnError(*sub, res.error(), callsite);
122+
} else {
123+
auto emplaceRes = sub->subgroups.emplace(identifier_, res.value());
124+
subgroupConsumerIt = emplaceRes.first;
125+
anyForwarded = true;
126+
fn(sub, subgroupConsumerIt->second);
116127
}
117128
}
118129
});
@@ -384,21 +395,26 @@ bool MoQForwarder::checkRange(const Subscriber& sub) {
384395
if (*largest_ < sub.range.start) {
385396
// future
386397
return false;
387-
} else if (*largest_ > sub.range.end) {
388-
// now past, send publishDone
389-
// TOOD: maybe this is too early for a relay.
390-
XLOG(DBG4) << "removeSubscriber from checkRange";
398+
}
399+
// TOOD: maybe sending publishDone here is too early for a relay.
400+
return !checkPastEnd(sub);
401+
}
402+
403+
bool MoQForwarder::checkPastEnd(const Subscriber& sub) {
404+
XCHECK(largest_);
405+
if (*largest_ > sub.range.end) {
406+
XLOG(DBG4) << "removeSubscriber from checkPastEnd";
391407
removeSubscriber(
392408
sub.session,
393409
PublishDone{
394410
sub.requestID,
395411
PublishDoneStatusCode::SUBSCRIPTION_ENDED,
396412
0, // filled in by session
397413
""},
398-
"checkRange");
399-
return false;
414+
"checkPastEnd");
415+
return true;
400416
}
401-
return true;
417+
return false;
402418
}
403419

404420
void MoQForwarder::removeSubscriberOnError(

‎moxygen/relay/MoQForwarder.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -194,6 +194,11 @@ class MoQForwarder : public TrackConsumer {
194194

195195
bool checkRange(const Subscriber& sub);
196196

197+
// Returns true if largest_ has advanced past sub.range.end. As a side
198+
// effect this also publishDone's the subscriber; that retirement
199+
// probably belongs elsewhere (TODO).
200+
bool checkPastEnd(const Subscriber& sub);
201+
197202
void removeSubscriberOnError(
198203
const Subscriber& sub,
199204
const MoQPublishError& err,

‎moxygen/relay/test/MoQForwarderTest.cpp‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1016,4 +1016,84 @@ TEST_F(MoQForwarderTest, RemoveForwardOnlySubscriberWithPublishDone) {
10161016
EXPECT_TRUE(forwarder->empty());
10171017
}
10181018

1019+
// Repro for moxygen#168: a PUBLISH_OK arriving after beginObject has already
1020+
// fanned out must not orphan the subscriber's open SubgroupConsumer.
1021+
// Sequence:
1022+
// 1. Subscribe (LargestObject -> range.start resolves to {0,0} with no data).
1023+
// 2. Publisher begins multi-chunk object 5; forwarder creates sg and
1024+
// delivers the initial payload.
1025+
// 3. PUBLISH_OK arrives late (LargestObject) -> range.start becomes {0,6}.
1026+
// 4. Publisher sends the object 5 continuation. Before the fix, the
1027+
// forwarder re-checked range.start on the existing sg and silently
1028+
// skipped it, stranding it with a partial object and tripping
1029+
// validatePublish on the next beginObject downstream.
1030+
TEST_F(MoQForwarderTest, SubscriberOnPublishOkDoesNotStrandPartialObject) {
1031+
auto subscriberSession = createMockSession();
1032+
auto mockConsumer = createMockConsumer();
1033+
1034+
constexpr uint64_t kObjectID = 5;
1035+
constexpr uint64_t kObjectLength = 100;
1036+
constexpr uint64_t kInitialLength = 20;
1037+
std::string downstreamPayload;
1038+
1039+
EXPECT_CALL(*mockConsumer, beginSubgroup(_, _, _, _))
1040+
.WillOnce([&](uint64_t, uint64_t, uint8_t, bool) {
1041+
auto sg = createMockSubgroupConsumer();
1042+
EXPECT_CALL(*sg, beginObject(kObjectID, kObjectLength, _, _))
1043+
.WillOnce([&](uint64_t, uint64_t, Payload p, Extensions) {
1044+
downstreamPayload += p->moveToFbString().toStdString();
1045+
return folly::makeExpected<MoQPublishError>(folly::unit);
1046+
});
1047+
EXPECT_CALL(*sg, objectPayload(_, _)).WillOnce([&](Payload p, bool) {
1048+
downstreamPayload += p->moveToFbString().toStdString();
1049+
return folly::makeExpected<MoQPublishError>(
1050+
ObjectPublishStatus::IN_PROGRESS);
1051+
});
1052+
return folly::
1053+
makeExpected<MoQPublishError, std::shared_ptr<SubgroupConsumer>>(
1054+
sg);
1055+
});
1056+
1057+
auto forwarder = std::make_shared<MoQForwarder>(kFwdTestTrackName);
1058+
1059+
auto subscriber =
1060+
addSubscriber(*forwarder, subscriberSession, mockConsumer, RequestID(1));
1061+
ASSERT_NE(subscriber, nullptr);
1062+
1063+
auto subgroupRes = forwarder->beginSubgroup(0, 0, 0);
1064+
ASSERT_TRUE(subgroupRes.hasValue());
1065+
auto subgroup = *subgroupRes;
1066+
1067+
auto initial = folly::IOBuf::copyBuffer(std::string(kInitialLength, 'a'));
1068+
ASSERT_TRUE(
1069+
subgroup->beginObject(kObjectID, kObjectLength, std::move(initial), {})
1070+
.hasValue());
1071+
1072+
// Late PUBLISH_OK with LargestObject: range.start becomes largest.object + 1.
1073+
PublishOk pubOk{
1074+
RequestID(1),
1075+
true, // forward
1076+
0, // subscriberPriority
1077+
GroupOrder::OldestFirst,
1078+
LocationType::LargestObject,
1079+
std::nullopt, // start
1080+
std::nullopt, // endGroup
1081+
TrackRequestParameters(FrameType::PUBLISH_OK)};
1082+
subscriber->onPublishOk(pubOk);
1083+
EXPECT_EQ(subscriber->range.start.object, kObjectID + 1);
1084+
1085+
auto continuation = folly::IOBuf::copyBuffer(
1086+
std::string(kObjectLength - kInitialLength, 'b'));
1087+
ASSERT_TRUE(
1088+
subgroup->objectPayload(std::move(continuation), /*finStream=*/false)
1089+
.hasValue());
1090+
1091+
ASSERT_TRUE(subgroup->endOfSubgroup().hasValue());
1092+
1093+
EXPECT_EQ(
1094+
downstreamPayload,
1095+
std::string(kInitialLength, 'a') +
1096+
std::string(kObjectLength - kInitialLength, 'b'));
1097+
}
1098+
10191099
} // namespace moxygen::test

0 commit comments

Comments
 (0)