Fix descending fetch gap-range bounds and tail-mark

Summary:
Three bugs in the descending-fetch path of FetchWriteback / getGapRanges, found by review of the LocationIntervalSet conversion:

1. getGapRanges DESC range #1 extended to {startGroup, MAX} regardless of the user's fetch end, marking positions above fetchEnd as nonexistent. Likewise range #3 started at {endGroup, 0} regardless of fetchStart. Either could shadow positions outside the requested range and suppress later upstream fetches. Fixed by adding fetchStart/fetchEnd parameters and clamping ranges #1 and #3 against them when start.group / end.group is the fetch boundary.

2. cacheImpl streaming objectPayload finFetch path called markNonExistentTo(end_) without first advancing the iterator past the just-completed object, so the gap range overlapped it. Fixed by stepping fetchRangeIt_.next() before tail-marking.

3. Both finFetch tail-marks called markNonExistentTo(end_), where end_ is the user's highest endpoint. In DESC iteration, the iteration end is the lowest position, so the wrong endpoint was used. Fixed by switching to fetchRangeIt_.end(), which returns the order-aware iteration end.

Adds three regression tests covering each bug.

Differential Revision: D103210059

Author: afrind

moxygen/relay/MoQCache.cpp

@@ -131,8 +131,16 @@ bool exists(ObjectStatus status) {
 // Helper to compute gap ranges for markNonExistentTo.
 // Returns vector of (start, end) intervals to mark as gaps.
 // Handles both ascending and descending iteration orders.
-std::vector<std::pair<AbsoluteLocation, AbsoluteLocation>>
-getGapRanges(AbsoluteLocation start, AbsoluteLocation end, GroupOrder order) {
+//
+// fetchStart/fetchEnd are the user's original fetch range (inclusive start,
+// exclusive end), used to clamp DESC top-group and bottom-group ranges so we
+// never mark positions outside the requested range as non-existent.
+std::vector<std::pair<AbsoluteLocation, AbsoluteLocation>> getGapRanges(
+    AbsoluteLocation start,
+    AbsoluteLocation end,
+    GroupOrder order,
+    AbsoluteLocation fetchStart,
+    AbsoluteLocation fetchEnd) {
   std::vector<std::pair<AbsoluteLocation, AbsoluteLocation>> ranges;
 
   // Check if range is empty based on iteration order
@@ -167,10 +175,13 @@ getGapRanges(AbsoluteLocation start, AbsoluteLocation end, GroupOrder order) {
   // Descending: groups are iterated in reverse, but objects within groups
   // are still ascending. This may require up to 3 ranges.
 
-  // 1. Rest of current group (start group in iteration, but higher group
-  // number)
-  ranges.emplace_back(
-      start, AbsoluteLocation{start.group, kLocationMax.object});
+  // 1. Rest of start.group, clamped to fetchEnd when start.group is the
+  // fetch's top group. start <= startGroupEnd is guaranteed by the
+  // same-group early return above and the invariant start <= fetchEnd-1.
+  AbsoluteLocation startGroupEnd = (start.group == fetchEnd.group)
+      ? *fetchEnd.prevInGroup()
+      : AbsoluteLocation{start.group, kLocationMax.object};
+  ranges.emplace_back(start, startGroupEnd);
 
   // 2. Intermediate groups (between start.group-1 and end.group+1)
   // start.group > end.group guaranteed (descending, different groups),
@@ -182,10 +193,16 @@ getGapRanges(AbsoluteLocation start, AbsoluteLocation end, GroupOrder order) {
     ranges.emplace_back(*endNextGroup, *startPrevGroupEnd);
   } // else the groups were consecutive
 
-  // 3. Partial target group (lower group number, objects before target.object)
-  if (auto prev = end.prevInGroup()) {
-    ranges.emplace_back(AbsoluteLocation{end.group, 0}, *prev);
-  } // else end.object == 0, no partial group to cover
+  // 3. Partial end.group, clamped to fetchStart when end.group is the fetch's
+  // bottom group. endGroupStart < end skips both end.object == 0 (no partial
+  // group) and end == fetchStart (iterator at its terminal position), and
+  // proves end.prevInGroup() is non-empty.
+  AbsoluteLocation endGroupStart = (end.group == fetchStart.group)
+      ? fetchStart
+      : AbsoluteLocation{end.group, 0};
+  if (endGroupStart < end) {
+    ranges.emplace_back(endGroupStart, *end.prevInGroup());
+  }
 
   return ranges;
 }
@@ -922,7 +939,7 @@ class MoQCache::FetchWriteback : public FetchConsumer {
             ? end
             : AbsoluteLocation{start.group, kLocationMax.object},
         this);
-    inProgressItersList_.push_back(setIt);
+    inProgressItersList_.emplace_back(start.group, setIt);
     fetchRangeIt_.track->activeFetchCount++;
 
     // Handle middle groups (if any)
@@ -931,14 +948,14 @@ class MoQCache::FetchWriteback : public FetchConsumer {
           AbsoluteLocation{currGroup, 0},
           AbsoluteLocation{currGroup, kLocationMax.object},
           this);
-      inProgressItersList_.push_back(setIt);
+      inProgressItersList_.emplace_back(currGroup, setIt);
       fetchRangeIt_.track->activeFetchCount++;
     }
     // Handle end group (if different from start)
     if (end.group != start.group) {
       setIt = fetchRangeIt_.track->fetchInProgress.insert(
           AbsoluteLocation{end.group, 0}, end, this);
-      inProgressItersList_.push_back(setIt);
+      inProgressItersList_.emplace_back(end.group, setIt);
       fetchRangeIt_.track->activeFetchCount++;
     }
 
@@ -952,8 +969,8 @@ class MoQCache::FetchWriteback : public FetchConsumer {
     inProgress_.post();
     if (!inProgressItersList_.empty()) {
       while (dualIter_ != dualIter_.end()) {
-        auto it = *dualIter_;
-        fetchRangeIt_.track->fetchInProgress.erase(it->start.group, it);
+        auto [groupKey, it] = *dualIter_;
+        fetchRangeIt_.track->fetchInProgress.erase(groupKey, it);
         fetchRangeIt_.track->activeFetchCount--;
         ++dualIter_;
       }
@@ -971,22 +988,22 @@ class MoQCache::FetchWriteback : public FetchConsumer {
     if (dualIter_ == dualIter_.end()) {
       return;
     }
+    auto& [groupKey, it] = *dualIter_;
     // iterators in dualIter_ are group scoped
-    if (start_.group == (*dualIter_)->start.group &&
-        start_ < (*dualIter_)->end) {
+    if (start_.group == groupKey && start_ < it->end) {
       // Update the start_ value of this interval
-      (*dualIter_)->start = start_;
+      it->start = start_;
       inProgress_.reset();
     } else {
       // Remove the iterator from track level tracking
-      fetchRangeIt_.track->fetchInProgress.erase(
-          (*dualIter_)->start.group, (*dualIter_));
+      fetchRangeIt_.track->fetchInProgress.erase(groupKey, it);
       ++dualIter_;
 
       // Iterator has processed the last element
       if (dualIter_ != dualIter_.end()) {
         // Update the start_ value of this interval
-        (*dualIter_)->start = *fetchRangeIt_;
+        auto& [nextGroupKey, nextIt] = *dualIter_;
+        nextIt->start = *fetchRangeIt_;
         inProgress_.reset();
       }
     }
@@ -1102,7 +1119,11 @@ class MoQCache::FetchWriteback : public FetchConsumer {
     cache_.totalCachedBytes_ += addedBytes;
     cache_.evictForByteLimitIfNeeded();
     if (finFetch) {
-      markNonExistentTo(end_);
+      // Iterator still ON the just-completed object; step past it before
+      // tail-marking so the gap range doesn't overlap it. Use the iterator's
+      // order-aware end (DESC: lowest position; ASC: end_).
+      fetchRangeIt_.next();
+      markNonExistentTo(fetchRangeIt_.end());
       updateInProgress();
     }
     return consumer_->objectPayload(std::move(payload), finFetch && proxyFin_);
@@ -1207,7 +1228,7 @@ class MoQCache::FetchWriteback : public FetchConsumer {
   AbsoluteLocation end_;
   bool proxyFin_{false};
   std::shared_ptr<FetchConsumer> consumer_;
-  std::vector<FetchInProgressSet::IntervalList::iterator> inProgressItersList_;
+  std::vector<MoQCache::InProgressFetchEntry> inProgressItersList_;
   MoQCache::InProgressFetchesIter dualIter_;
   folly::coro::Baton inProgress_;
   folly::coro::Baton complete_;
@@ -1225,7 +1246,12 @@ class MoQCache::FetchWriteback : public FetchConsumer {
       return;
     }
 
-    auto ranges = getGapRanges(*fetchRangeIt_, target, fetchRangeIt_.order);
+    auto ranges = getGapRanges(
+        *fetchRangeIt_,
+        target,
+        fetchRangeIt_.order,
+        fetchRangeIt_.minLocation,
+        fetchRangeIt_.maxLocation);
     for (const auto& [start, end] : ranges) {
       fetchRangeIt_.track->gaps.insert(start, end);
     }
@@ -1274,7 +1300,10 @@ class MoQCache::FetchWriteback : public FetchConsumer {
       fetchRangeIt_.next();
       updateInProgress();
       if (finFetch) {
-        markNonExistentTo(end_);
+        // Use the iterator's order-aware end. In DESC, end_ is the user's
+        // highest endpoint (the wrong direction); fetchRangeIt_.end()
+        // returns the iteration end (the lowest position).
+        markNonExistentTo(fetchRangeIt_.end());
         complete_.post();
       }
     }

moxygen/relay/MoQCache.h

@@ -209,9 +209,10 @@ class MoQCache {
 
   // Entry for a track
   using FetchInProgressSet = FetchIntervalSet<FetchWriteback*>;
+  using InProgressFetchEntry =
+      std::pair<uint64_t, FetchInProgressSet::IntervalList::iterator>;
   // Type alias for the complex BidiIterator type used in FetchWriteback
-  using InProgressFetchesIter =
-      BidiIterator<std::vector<FetchInProgressSet::IntervalList::iterator>>;
+  using InProgressFetchesIter = BidiIterator<std::vector<InProgressFetchEntry>>;
 
   struct CacheTrack {
     folly::F14FastMap<uint64_t, std::shared_ptr<CacheGroup>> groups;

moxygen/relay/test/MoQCacheTests.cpp

@@ -2047,6 +2047,188 @@ CO_TEST_F(MoQCacheTest, TestDescendingGapNearGroupZero) {
   EXPECT_TRUE(res.hasValue());
 }
 
+// Regression: DESC getGapRanges range #1 used to extend to {topGroup, MAX}
+// instead of clamping to fetchEnd, so a cold DESC fetch could mark positions
+// above fetchEnd.object as nonexistent and suppress later upstream fetches.
+CO_TEST_F(MoQCacheTest, TestDescendingFetchDoesNotMarkTopGroupTailAsGap) {
+  auto firstConsumer = std::make_shared<NiceMock<MockFetchConsumer>>();
+  ON_CALL(*firstConsumer, object(_, _, _, _, _, _, _))
+      .WillByDefault(Return(folly::unit));
+  ON_CALL(*firstConsumer, endOfFetch()).WillByDefault(Return(folly::unit));
+  EXPECT_CALL(*upstream_, fetch(_, _))
+      .WillOnce([this](Fetch fetch, std::shared_ptr<FetchConsumer> consumer) {
+        auto [standalone, joining] = fetchType(fetch);
+        EXPECT_EQ(standalone->start, (AbsoluteLocation{3, 3}));
+        EXPECT_EQ(standalone->end, (AbsoluteLocation{5, 10}));
+        EXPECT_EQ(fetch.groupOrder, GroupOrder::NewestFirst);
+        auto res =
+            consumer->object(3, 0, 3, makeBuf(100), noExtensions(), true);
+        EXPECT_FALSE(res.hasError());
+        res = consumer->endOfFetch();
+        EXPECT_FALSE(res.hasError());
+        upstreamFetchHandle_ =
+            std::make_shared<moxygen::MockFetchHandle>(FetchOk{
+                0,
+                GroupOrder::NewestFirst,
+                false,
+                AbsoluteLocation{3, 3},
+            });
+        return folly::coro::makeTask<Publisher::FetchResult>(
+            upstreamFetchHandle_);
+      })
+      .RetiresOnSaturation();
+
+  auto res = co_await cache_.fetch(
+      getFetch({3, 3}, {5, 10}, GroupOrder::NewestFirst),
+      firstConsumer,
+      upstream_);
+  EXPECT_TRUE(res.hasValue());
+
+  // {5, 20} was outside the original fetch range; cache must reach upstream.
+  auto laterConsumer = std::make_shared<NiceMock<MockFetchConsumer>>();
+  ON_CALL(*laterConsumer, endOfFetch()).WillByDefault(Return(folly::unit));
+  EXPECT_CALL(*upstream_, fetch(_, _))
+      .WillOnce([this](Fetch fetch, std::shared_ptr<FetchConsumer> consumer) {
+        auto [standalone, joining] = fetchType(fetch);
+        EXPECT_EQ(standalone->start, (AbsoluteLocation{5, 20}));
+        EXPECT_EQ(standalone->end, (AbsoluteLocation{5, 21}));
+        EXPECT_EQ(fetch.groupOrder, GroupOrder::OldestFirst);
+        auto res = consumer->endOfFetch();
+        EXPECT_FALSE(res.hasError());
+        upstreamFetchHandle_ =
+            std::make_shared<moxygen::MockFetchHandle>(FetchOk{
+                0,
+                GroupOrder::OldestFirst,
+                false,
+                AbsoluteLocation{5, 20},
+            });
+        return folly::coro::makeTask<Publisher::FetchResult>(
+            upstreamFetchHandle_);
+      })
+      .RetiresOnSaturation();
+
+  res = co_await cache_.fetch(
+      getFetch({5, 20}, {5, 21}, GroupOrder::OldestFirst),
+      laterConsumer,
+      upstream_);
+  EXPECT_TRUE(res.hasValue());
+}
+
+// Regression: DESC getGapRanges range #3 used to start at {bottomGroup, 0}
+// instead of clamping to fetchStart.object, shadowing positions below the
+// requested lower bound.
+CO_TEST_F(MoQCacheTest, TestDescendingFetchDoesNotMarkBottomGroupHeadAsGap) {
+  auto firstConsumer = std::make_shared<NiceMock<MockFetchConsumer>>();
+  ON_CALL(*firstConsumer, object(_, _, _, _, _, _, _))
+      .WillByDefault(Return(folly::unit));
+  ON_CALL(*firstConsumer, endOfFetch()).WillByDefault(Return(folly::unit));
+  EXPECT_CALL(*upstream_, fetch(_, _))
+      .WillOnce([this](Fetch fetch, std::shared_ptr<FetchConsumer> consumer) {
+        auto [standalone, joining] = fetchType(fetch);
+        EXPECT_EQ(standalone->start, (AbsoluteLocation{3, 5}));
+        EXPECT_EQ(standalone->end, (AbsoluteLocation{5, 10}));
+        EXPECT_EQ(fetch.groupOrder, GroupOrder::NewestFirst);
+        auto res =
+            consumer->object(3, 0, 5, makeBuf(100), noExtensions(), true);
+        EXPECT_FALSE(res.hasError());
+        res = consumer->endOfFetch();
+        EXPECT_FALSE(res.hasError());
+        upstreamFetchHandle_ =
+            std::make_shared<moxygen::MockFetchHandle>(FetchOk{
+                0,
+                GroupOrder::NewestFirst,
+                false,
+                AbsoluteLocation{3, 5},
+            });
+        return folly::coro::makeTask<Publisher::FetchResult>(
+            upstreamFetchHandle_);
+      })
+      .RetiresOnSaturation();
+
+  auto res = co_await cache_.fetch(
+      getFetch({3, 5}, {5, 10}, GroupOrder::NewestFirst),
+      firstConsumer,
+      upstream_);
+  EXPECT_TRUE(res.hasValue());
+
+  // {3, 1} was below fetchStart.object=5; cache must reach upstream.
+  auto laterConsumer = std::make_shared<NiceMock<MockFetchConsumer>>();
+  ON_CALL(*laterConsumer, endOfFetch()).WillByDefault(Return(folly::unit));
+  EXPECT_CALL(*upstream_, fetch(_, _))
+      .WillOnce([this](Fetch fetch, std::shared_ptr<FetchConsumer> consumer) {
+        auto [standalone, joining] = fetchType(fetch);
+        EXPECT_EQ(standalone->start, (AbsoluteLocation{3, 1}));
+        EXPECT_EQ(standalone->end, (AbsoluteLocation{3, 2}));
+        auto res = consumer->endOfFetch();
+        EXPECT_FALSE(res.hasError());
+        upstreamFetchHandle_ =
+            std::make_shared<moxygen::MockFetchHandle>(FetchOk{
+                0,
+                GroupOrder::OldestFirst,
+                false,
+                AbsoluteLocation{3, 1},
+            });
+        return folly::coro::makeTask<Publisher::FetchResult>(
+            upstreamFetchHandle_);
+      })
+      .RetiresOnSaturation();
+
+  res = co_await cache_.fetch(
+      getFetch({3, 1}, {3, 2}, GroupOrder::OldestFirst),
+      laterConsumer,
+      upstream_);
+  EXPECT_TRUE(res.hasValue());
+}
+
+// Regression: DESC finFetch tail-marked against raw end_ (the user's highest
+// endpoint) instead of the iterator's order-aware end. Combined with the
+// off-by-one in cacheImpl, this could shadow the just-cached object.
+CO_TEST_F(MoQCacheTest, TestDescendingFetchDoesNotGapCachedObject) {
+  auto firstConsumer = std::make_shared<NiceMock<MockFetchConsumer>>();
+  ON_CALL(*firstConsumer, object(_, _, _, _, _, _, _))
+      .WillByDefault(Return(folly::unit));
+  ON_CALL(*firstConsumer, endOfFetch()).WillByDefault(Return(folly::unit));
+  EXPECT_CALL(*upstream_, fetch(_, _))
+      .WillOnce([this](Fetch fetch, std::shared_ptr<FetchConsumer> consumer) {
+        auto [standalone, joining] = fetchType(fetch);
+        EXPECT_EQ(standalone->start, (AbsoluteLocation{3, 3}));
+        EXPECT_EQ(standalone->end, (AbsoluteLocation{5, 10}));
+        EXPECT_EQ(fetch.groupOrder, GroupOrder::NewestFirst);
+        auto res =
+            consumer->object(3, 0, 3, makeBuf(100), noExtensions(), true);
+        EXPECT_FALSE(res.hasError());
+        res = consumer->endOfFetch();
+        EXPECT_FALSE(res.hasError());
+        upstreamFetchHandle_ =
+            std::make_shared<moxygen::MockFetchHandle>(FetchOk{
+                0,
+                GroupOrder::NewestFirst,
+                false,
+                AbsoluteLocation{3, 3},
+            });
+        return folly::coro::makeTask<Publisher::FetchResult>(
+            upstreamFetchHandle_);
+      })
+      .RetiresOnSaturation();
+
+  auto res = co_await cache_.fetch(
+      getFetch({3, 3}, {5, 10}, GroupOrder::NewestFirst),
+      firstConsumer,
+      upstream_);
+  EXPECT_TRUE(res.hasValue());
+
+  // The cached object at {3, 3} must remain servable from cache.
+  auto serveConsumer = std::make_shared<NiceMock<MockFetchConsumer>>();
+  ON_CALL(*serveConsumer, endOfFetch()).WillByDefault(Return(folly::unit));
+  EXPECT_CALL(*serveConsumer, object(3, 0, 3, _, _, true, _))
+      .WillOnce(Return(folly::unit));
+  res = co_await cache_.fetch(
+      getFetch({3, 3}, {3, 4}, GroupOrder::OldestFirst),
+      serveConsumer,
+      upstream_);
+  EXPECT_TRUE(res.hasValue());
+}
+
 TEST_F(MoQCacheTest, TestForwardingPreferenceMismatchIsMalformedTrack) {
   // If an object is cached with one forwarding preference and we try to cache
   // the same object with a different forwarding preference, it should fail