diff --git a/moxygen/MoQTypes.h b/moxygen/MoQTypes.h index 4e890df65..746085fa2 100644 --- a/moxygen/MoQTypes.h +++ b/moxygen/MoQTypes.h @@ -343,6 +343,89 @@ struct AbsoluteLocation { } std::string describe() const; + + /** + * Get the successor location (next position in lexicographic order). + * Returns nullopt if this is kLocationMax. + */ + std::optional next() const { + if (group == kEightByteLimit && object == kEightByteLimit) { + return std::nullopt; + } + if (object == kEightByteLimit) { + return AbsoluteLocation{group + 1, 0}; + } + return AbsoluteLocation{group, object + 1}; + } + + /** + * Get the predecessor location. + * Returns nullopt if this is kLocationMin. + */ + std::optional prev() const { + if (group == 0 && object == 0) { + return std::nullopt; + } + if (object == 0) { + return AbsoluteLocation{group - 1, kEightByteLimit}; + } + return AbsoluteLocation{group, object - 1}; + } + + /** + * Get the next object within the same group. + * Returns nullopt if object is at max. + */ + std::optional nextInGroup() const { + if (object == kEightByteLimit) { + return std::nullopt; + } + return AbsoluteLocation{group, object + 1}; + } + + /** + * Get the start of the next group {group+1, 0}. + * Returns nullopt if group is at max. + */ + std::optional nextGroup() const { + if (group == kEightByteLimit) { + return std::nullopt; + } + return AbsoluteLocation{group + 1, 0}; + } + + /** + * Get the start of the previous group {group-1, 0}. + * Returns nullopt if group is 0. + */ + std::optional prevGroup() const { + if (group == 0) { + return std::nullopt; + } + return AbsoluteLocation{group - 1, 0}; + } + + /** + * Get the last location in the previous group {group-1, MAX}. + * Returns nullopt if group is 0. + */ + std::optional prevGroupEnd() const { + if (group == 0) { + return std::nullopt; + } + return AbsoluteLocation{group - 1, kEightByteLimit}; + } + + /** + * Get the previous object within the same group. + * Returns nullopt if object is 0. + */ + std::optional prevInGroup() const { + if (object == 0) { + return std::nullopt; + } + return AbsoluteLocation{group, object - 1}; + } }; constexpr AbsoluteLocation kLocationMin; diff --git a/moxygen/relay/CMakeLists.txt b/moxygen/relay/CMakeLists.txt index d7762b3c1..b96528105 100644 --- a/moxygen/relay/CMakeLists.txt +++ b/moxygen/relay/CMakeLists.txt @@ -19,6 +19,7 @@ moxygen_add_library(moxygen_relay_moq_cache SRCS MoQCache.cpp EXPORTED_DEPS + moxygen_util_location_interval_set moxygen_moq moxygen_moq_consumers moxygen_moq_framer diff --git a/moxygen/relay/MoQCache.cpp b/moxygen/relay/MoQCache.cpp index 4f2578780..1ad1dc4ef 100644 --- a/moxygen/relay/MoQCache.cpp +++ b/moxygen/relay/MoQCache.cpp @@ -15,6 +15,36 @@ namespace { using namespace moxygen; +// Cap on prior-gap validation scans to avoid O(n) work for huge gaps. +constexpr uint64_t kMaxGapValidation = 100; + +// Compute the next iterator position after a gap when iterating descending +// (NewestFirst). Groups iterate high-to-low, but objects within a group go +// low-to-high. Returns std::nullopt if the gap extends past the iteration +// boundary. +std::optional computeDescendingAfterGap( + const AbsoluteLocation& current, + const AbsoluteLocation& gapStart, + const AbsoluteLocation& minLocation) { + // Gap starts mid-group in a DIFFERENT group: there may be valid objects + // before the gap in that group, so jump to the start of it. + if (gapStart.object > 0 && gapStart.group != current.group) { + AbsoluteLocation afterGap{gapStart.group, 0}; + if (afterGap.group == minLocation.group) { + afterGap.object = minLocation.object; + } + return afterGap; + } + // Either gap starts at object 0, OR gap starts in the current group. If gap + // starts in the current group, we've already served objects before the gap + // (since objects iterate low-to-high). Skip to the previous group. + auto afterGap = gapStart.prevGroup(); + if (afterGap && afterGap->group == minLocation.group) { + afterGap->object = minLocation.object; + } + return afterGap; +} + // Splits one contiguous cache miss region into multiple intervals // Depending on whether we are fetching asc or desc // For asc, no split. Desc cases may or may not need splitting @@ -92,8 +122,98 @@ bool isEndOfTrack(ObjectStatus status) { } bool exists(ObjectStatus status) { - return status != ObjectStatus::OBJECT_NOT_EXIST && - status != ObjectStatus::GROUP_NOT_EXIST; + // With NOT_EXIST statuses removed, all valid statuses represent real objects + return status == ObjectStatus::NORMAL || + status == ObjectStatus::END_OF_GROUP || + status == ObjectStatus::END_OF_TRACK; +} + +// Helper to compute gap ranges for markNonExistentTo. +// Returns vector of (start, end) intervals to mark as gaps. +// Handles both ascending and descending iteration orders. +// +// 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> getGapRanges( + AbsoluteLocation start, + AbsoluteLocation end, + GroupOrder order, + AbsoluteLocation fetchStart, + AbsoluteLocation fetchEnd) { + std::vector> ranges; + + // Check if range is empty based on iteration order + // Ascending: start should be < end + // Descending: groups go high-to-low, but objects within groups go low-to-high + // so start.group > end.group OR (same group with start.object < end.object) + bool isEmpty = (order != GroupOrder::NewestFirst) + ? (start >= end) + : (start.group < end.group || + (start.group == end.group && start.object >= end.object)); + if (isEmpty) { + return ranges; + } + + if (start.group == end.group) { + // Same group - simple range from start to end-1 + if (auto prev = end.prevInGroup()) { + ranges.emplace_back(start, *prev); + } + return ranges; + } + + // For ascending: straightforward contiguous range + if (order != GroupOrder::NewestFirst) { + auto gapEnd = end.prev(); + XCHECK(gapEnd) + << "ascending different-group with end={0,0} should be impossible"; + ranges.emplace_back(start, *gapEnd); + return ranges; + } + + // Descending: groups are iterated in reverse, but objects within groups + // are still ascending. This may require up to 3 ranges. + + // 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), + // so end.group < MAX and start.group > 0. + auto endNextGroup = end.nextGroup(); + auto startPrevGroupEnd = start.prevGroupEnd(); + XCHECK(endNextGroup && startPrevGroupEnd); + if (startPrevGroupEnd->group >= endNextGroup->group) { + ranges.emplace_back(*endNextGroup, *startPrevGroupEnd); + } // else the groups were consecutive + + // 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; +} + +// Check if a group is known to be entirely nonexistent (all locations in +// [{groupID, 0}, {groupID, MAX}] are in gaps). +bool isGroupNonExistent(const LocationIntervalSet& gaps, uint64_t groupID) { + auto end = gaps.findIntervalEnd({groupID, 0}); + return end && + (end->group > groupID || + (end->group == groupID && end->object == kLocationMax.object)); } folly::Expected publishObject( @@ -112,13 +232,12 @@ folly::Expected publishObject( object.extensions, lastObject, object.forwardingPreferenceIsDatagram); - // These are implicit + // These are invalid statuses - should not be in the cache case ObjectStatus::OBJECT_NOT_EXIST: case ObjectStatus::GROUP_NOT_EXIST: - if (lastObject) { - consumer->endOfFetch(); - } - return folly::unit; + return folly::makeUnexpected( + MoQPublishError{ + MoQPublishError::API_ERROR, "Invalid NOT_EXIST status"}); case ObjectStatus::END_OF_GROUP: return consumer->endOfGroup( current.group, object.subgroup, current.object, lastObject); @@ -135,6 +254,8 @@ folly::Expected publishObject( namespace moxygen { folly::Expected MoQCache::CacheGroup::cacheObject( + CacheTrack& track, + uint64_t groupID, uint64_t subgroup, uint64_t objectID, ObjectStatus status, @@ -143,20 +264,62 @@ folly::Expected MoQCache::CacheGroup::cacheObject( bool complete, bool forwardingPreferenceIsDatagram, TimePoint now) { - XLOG(DBG1) << "caching objID=" << objectID << " status=" << (uint32_t)status + XLOG(DBG1) << "caching group=" << groupID << " objID=" << objectID + << " status=" << (uint32_t)status << " complete=" << uint32_t(complete); + + // NOT_EXIST statuses are invalid - they shouldn't be passed to cacheObject + if (status == ObjectStatus::OBJECT_NOT_EXIST || + status == ObjectStatus::GROUP_NOT_EXIST) { + XLOG(ERR) << "Invalid NOT_EXIST status passed to cacheObject; objID=" + << objectID; + return folly::makeUnexpected(MoQPublishError( + MoQPublishError::API_ERROR, "Invalid NOT_EXIST status")); + } + + // Reject caching into a known gap + if (track.gaps.contains({groupID, objectID})) { + XLOG(ERR) << "Attempting to cache object in known gap; group=" << groupID + << " objID=" << objectID; + return folly::makeUnexpected(MoQPublishError( + MoQPublishError::MALFORMED_TRACK, "Object in known gap")); + } + + // Process gap extensions first + auto gapRes = track.processGapExtensions(groupID, objectID, extensions); + if (gapRes.hasError()) { + return gapRes; + } + + // Handle END_OF_GROUP and END_OF_TRACK - mark remaining objects/groups as + // gaps. TODO: we could skip caching EOG/EOT entirely by including objectID + // in the gap range (start at objectID instead of objectID+1), then infer + // end-of-group/track from gap data when serving. + if (status == ObjectStatus::END_OF_GROUP || + status == ObjectStatus::END_OF_TRACK) { + if (auto next = AbsoluteLocation{groupID, objectID}.nextInGroup()) { + track.gaps.insert(*next, {groupID, kLocationMax.object}); + } + if (status == ObjectStatus::END_OF_TRACK) { + if (auto nextGrp = AbsoluteLocation{groupID, 0}.nextGroup()) { + track.gaps.insert(*nextGrp, kLocationMax); + } + } + endOfGroup = true; + } + + // Cache the object (including END_OF_GROUP/END_OF_TRACK status) // Compute new payload size once before the move size_t newPayloadSize = payload ? payload->computeChainDataLength() : 0; auto it = objects.find(objectID); if (it != objects.end()) { auto& cachedObject = it->second; - if (status != cachedObject->status && - status != ObjectStatus::OBJECT_NOT_EXIST) { + if (status != cachedObject->status) { XLOG(ERR) << "Invalid cache status change; objID=" << objectID << " status=" << (uint32_t)status << " already exists with different status"; - return folly::makeUnexpected( - MoQPublishError(MoQPublishError::API_ERROR, "Invalid status change")); + return folly::makeUnexpected(MoQPublishError( + MoQPublishError::MALFORMED_TRACK, "Invalid status change")); } if (status == ObjectStatus::NORMAL && cachedObject->complete && ((!payload && cachedObject->payload) || @@ -164,8 +327,8 @@ folly::Expected MoQCache::CacheGroup::cacheObject( (payload && cachedObject->payload && newPayloadSize != cachedObject->payloadSize))) { XLOG(ERR) << "Payload mismatch; objID=" << objectID; - return folly::makeUnexpected( - MoQPublishError(MoQPublishError::API_ERROR, "payload mismatch")); + return folly::makeUnexpected(MoQPublishError( + MoQPublishError::MALFORMED_TRACK, "payload mismatch")); } if (cachedObject->forwardingPreferenceIsDatagram != forwardingPreferenceIsDatagram) { @@ -174,6 +337,10 @@ folly::Expected MoQCache::CacheGroup::cacheObject( return folly::makeUnexpected(MoQPublishError( MoQPublishError::MALFORMED_TRACK, "forwardingPreference mismatch")); } + + // TODO: Consider removing status from CacheEntry. For fetch streams, we + // could only publish NORMAL objects and let END_OF_GROUP/END_OF_TRACK be + // implicit from the gap information. cachedObject->status = status; cachedObject->extensions = extensions; totalBytes -= cachedObject->payloadSize; @@ -197,35 +364,10 @@ folly::Expected MoQCache::CacheGroup::cacheObject( } if (objectID >= maxCachedObject) { maxCachedObject = objectID; - endOfGroup = - (status == ObjectStatus::END_OF_GROUP || - status == ObjectStatus::GROUP_NOT_EXIST); } return folly::unit; } -void MoQCache::CacheGroup::cacheMissingStatus( - uint64_t objectID, - ObjectStatus status) { - XLOG(DBG1) << "caching missing objID=" << objectID; - static constexpr auto kInvalidSubgroup = std::numeric_limits::max(); - // Always called with nullptr payload, so newPayloadSize == 0 and totalBytes - // will not increase. It also never decreases: callers in processGapExtensions - // guard against overwriting any existing object that has a real payload - // (GROUP_NOT_EXIST is only written to fresh gap groups; OBJECT_NOT_EXIST is - // only written to object IDs that are not yet in the cache). - // Use TimePoint::max() so gaps never expire - cacheObject( - kInvalidSubgroup, - objectID, - status, - noExtensions(), - nullptr, - true, - false, - TimePoint::max()); -} - class MoQCache::FetchHandle : public Publisher::FetchHandle { public: explicit FetchHandle(FetchOk ok) : Publisher::FetchHandle(std::move(ok)) {} @@ -341,22 +483,24 @@ MoQCache::CacheTrack::processGapExtensions( // First time seeing Prior Group ID Gap for this group currentGroup.seenPriorGroupIdGap = gap; - // Cache groups in the gap as GROUP_NOT_EXIST - // If groupID is G and gap is N, groups G-N to G-1 don't exist - for (uint64_t g = groupID - gap; g < groupID; ++g) { - auto& group = getOrCreateGroup(g); - // Allow if already marked as GROUP_NOT_EXIST (single object 0) - if (!group.objects.empty() && - (group.objects.size() != 1 || group.objects.begin()->first != 0 || - group.objects.begin()->second->status != - ObjectStatus::GROUP_NOT_EXIST)) { - XLOG(ERR) << "Prior Group ID Gap covers existing object in group " - << g; - return folly::makeUnexpected(MoQPublishError( - MoQPublishError::MALFORMED_TRACK, - "Prior Group ID Gap covers existing object")); + if (gap > 0) { + // Check for conflicts with existing real objects in the gap + // Only check up to kMaxGapValidation items to avoid O(n) for huge gaps + uint64_t checkCount = std::min(gap, kMaxGapValidation); + for (uint64_t g = groupID - checkCount; g < groupID; ++g) { + auto groupIt = groups.find(g); + if (groupIt != groups.end() && !groupIt->second->objects.empty()) { + XLOG(ERR) << "Prior Group ID Gap covers existing object in group " + << g; + return folly::makeUnexpected(MoQPublishError( + MoQPublishError::MALFORMED_TRACK, + "Prior Group ID Gap covers existing object")); + } } - group.cacheMissingStatus(0, ObjectStatus::GROUP_NOT_EXIST); + + // Mark groups in the gap as known (non-existent) via + // LocationIntervalSet + gaps.insert({groupID - gap, 0}, {groupID - 1, kLocationMax.object}); } } } @@ -379,24 +523,24 @@ MoQCache::CacheTrack::processGapExtensions( "Prior Object ID Gap larger than Object ID")); } - // Cache objects in the gap as OBJECT_NOT_EXIST - // If objectID is O and gap is N, objects O-N to O-1 don't exist - auto& group = getOrCreateGroup(groupID); - for (uint64_t o = objectID - gap; o < objectID; ++o) { - auto it = group.objects.find(o); - if (it != group.objects.end()) { - // Object already exists - only ok if it's already OBJECT_NOT_EXIST - if (it->second->status != ObjectStatus::OBJECT_NOT_EXIST) { + if (gap > 0) { + // Check for conflicts with existing real objects in the gap + // Only check up to kMaxGapValidation items to avoid O(n) for huge gaps + uint64_t checkCount = std::min(gap, kMaxGapValidation); + auto& group = getOrCreateGroup(groupID); + for (uint64_t o = objectID - checkCount; o < objectID; ++o) { + auto it = group.objects.find(o); + if (it != group.objects.end()) { XLOG(ERR) << "Prior Object ID Gap covers existing object " << o << " in group " << groupID; return folly::makeUnexpected(MoQPublishError( MoQPublishError::MALFORMED_TRACK, "Prior Object ID Gap covers existing object")); } - // Already marked as not existing, skip - continue; } - group.cacheMissingStatus(o, ObjectStatus::OBJECT_NOT_EXIST); + + // Mark objects in the gap as known (non-existent) via LocationIntervalSet + gaps.insert({groupID, objectID - gap}, {groupID, objectID - 1}); } } @@ -451,6 +595,8 @@ class MoQCache::SubgroupWriteback : public SubgroupConsumer { auto cPayload = payload ? payload->clone() : nullptr; auto cacheRes = cache_.cacheObjectAndUpdateBytes( cacheGroup_, + cacheTrack_, + group_, subgroup_, objID, ObjectStatus::NORMAL, @@ -463,11 +609,6 @@ class MoQCache::SubgroupWriteback : public SubgroupConsumer { consumer_->reset(ResetStreamErrorCode::INTERNAL_ERROR); return cacheRes; } - auto gapRes = cacheTrack_.processGapExtensions(group_, objID, ext); - if (gapRes.hasError()) { - consumer_->reset(ResetStreamErrorCode::INTERNAL_ERROR); - return gapRes; - } return consumer_->object(objID, std::move(payload), std::move(ext), finSub); } @@ -491,6 +632,8 @@ class MoQCache::SubgroupWriteback : public SubgroupConsumer { } auto cacheRes = cache_.cacheObjectAndUpdateBytes( cacheGroup_, + cacheTrack_, + group_, subgroup_, objectID, ObjectStatus::NORMAL, @@ -503,12 +646,6 @@ class MoQCache::SubgroupWriteback : public SubgroupConsumer { consumer_->reset(ResetStreamErrorCode::INTERNAL_ERROR); return cacheRes; } - auto gapRes = - cacheTrack_.processGapExtensions(group_, objectID, extensions); - if (gapRes.hasError()) { - consumer_->reset(ResetStreamErrorCode::INTERNAL_ERROR); - return gapRes; - } currentObject_ = objectID; currentLength_ = length; if (initialPayload) { @@ -555,8 +692,13 @@ class MoQCache::SubgroupWriteback : public SubgroupConsumer { consumer_->reset(ResetStreamErrorCode::INTERNAL_ERROR); return res; } + // TODO: END_OF_GROUP doesn't need to be cached as an object — just insert + // the gap. Requires removing status from CacheEntry and inferring + // end-of-group from gap data in publishObject. auto cacheRes = cache_.cacheObjectAndUpdateBytes( cacheGroup_, + cacheTrack_, + group_, subgroup_, endOfGroupObjectID, ObjectStatus::END_OF_GROUP, @@ -584,6 +726,8 @@ class MoQCache::SubgroupWriteback : public SubgroupConsumer { } auto cacheRes = cache_.cacheObjectAndUpdateBytes( cacheGroup_, + cacheTrack_, + group_, subgroup_, endOfTrackObjectID, ObjectStatus::END_OF_TRACK, @@ -663,6 +807,14 @@ class MoQCache::SubscribeWriteback : public TrackConsumer { Priority priority, bool /*containsLastInGroup*/ = false) override { // TODO: Handle containsLastInGroup parameter when caching + // Check if the group is known to not exist + if (isGroupNonExistent(track_.gaps, groupID)) { + XLOG(ERR) << "Attempting to begin subgroup in group already marked as " + "not existing; group=" + << groupID; + return folly::makeUnexpected(MoQPublishError( + MoQPublishError::MALFORMED_TRACK, "Invalid status change")); + } auto res = consumer_->beginSubgroup(groupID, subgroupID, priority); if (res.hasValue() && !track_.evicted) { track_.getOrCreateGroup(groupID, &cache_); @@ -698,6 +850,8 @@ class MoQCache::SubscribeWriteback : public TrackConsumer { } auto cacheRes = cache_.cacheObjectAndUpdateBytes( track_.getOrCreateGroup(header.group, &cache_), + track_, + header.group, header.subgroup, header.id, header.status, @@ -709,11 +863,6 @@ class MoQCache::SubscribeWriteback : public TrackConsumer { if (cacheRes.hasError()) { return cacheRes; } - auto gapRes = - track_.processGapExtensions(header.group, header.id, header.extensions); - if (gapRes.hasError()) { - return gapRes; - } return consumer_->objectStream(header, std::move(payload)); } @@ -732,6 +881,8 @@ class MoQCache::SubscribeWriteback : public TrackConsumer { } auto cacheRes = cache_.cacheObjectAndUpdateBytes( track_.getOrCreateGroup(header.group, &cache_), + track_, + header.group, header.subgroup, header.id, header.status, @@ -743,11 +894,6 @@ class MoQCache::SubscribeWriteback : public TrackConsumer { if (cacheRes.hasError()) { return cacheRes; } - auto gapRes = - track_.processGapExtensions(header.group, header.id, header.extensions); - if (gapRes.hasError()) { - return gapRes; - } return consumer_->datagram(header, std::move(payload)); } @@ -765,7 +911,7 @@ class MoQCache::SubscribeWriteback : public TrackConsumer { }; // Caches incoming objects and forwards them to the consumer. Handles gaps in -// the range by caching missing object status. +// the range by marking NonExistent objects. class MoQCache::FetchWriteback : public FetchConsumer { public: FetchWriteback( @@ -791,7 +937,7 @@ class MoQCache::FetchWriteback : public FetchConsumer { start, (start.group == end.group) ? end - : AbsoluteLocation{start.group, std::numeric_limits::max()}, + : AbsoluteLocation{start.group, kLocationMax.object}, this); inProgressItersList_.push_back(setIt); fetchRangeIt_.track->activeFetchCount++; @@ -800,7 +946,7 @@ class MoQCache::FetchWriteback : public FetchConsumer { for (auto currGroup = start.group + 1; currGroup < end.group; ++currGroup) { setIt = fetchRangeIt_.track->fetchInProgress.insert( AbsoluteLocation{currGroup, 0}, - AbsoluteLocation{currGroup, std::numeric_limits::max()}, + AbsoluteLocation{currGroup, kLocationMax.object}, this); inProgressItersList_.push_back(setIt); fetchRangeIt_.track->activeFetchCount++; @@ -873,10 +1019,6 @@ class MoQCache::FetchWriteback : public FetchConsumer { } } - void noObjects() { - cacheMissing(end_); - } - folly::Expected object( uint64_t gID, uint64_t sgID, @@ -909,10 +1051,6 @@ class MoQCache::FetchWriteback : public FetchConsumer { if (!res) { return res; } - auto gapRes = fetchRangeIt_.track->processGapExtensions(gID, objID, ext); - if (gapRes.hasError()) { - return gapRes; - } XLOG(DBG1) << "forward object " << AbsoluteLocation(gID, objID); return consumer_->object( gID, @@ -946,10 +1084,6 @@ class MoQCache::FetchWriteback : public FetchConsumer { if (!res) { return res; } - auto gapRes = fetchRangeIt_.track->processGapExtensions(gID, objID, ext); - if (gapRes.hasError()) { - return gapRes; - } currentLength_ = len; if (initPayload) { size_t initBytes = initPayload->computeChainDataLength(); @@ -985,7 +1119,11 @@ class MoQCache::FetchWriteback : public FetchConsumer { cache_.totalCachedBytes_ += addedBytes; cache_.evictForByteLimitIfNeeded(); if (finFetch) { - cacheMissing(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_); @@ -996,6 +1134,7 @@ class MoQCache::FetchWriteback : public FetchConsumer { if (fetchRangeIt_.track->evicted) { return consumer_->endOfGroup(gID, sgID, objID, fin && proxyFin_); } + // cacheImpl -> cacheObject inserts gap for remaining objects in this group constexpr auto kEndOfGroup = ObjectStatus::END_OF_GROUP; auto res = cacheImpl( gID, sgID, objID, kEndOfGroup, noExtensions(), nullptr, true, fin); @@ -1010,6 +1149,7 @@ class MoQCache::FetchWriteback : public FetchConsumer { if (fetchRangeIt_.track->evicted) { return consumer_->endOfTrackAndGroup(gID, sgID, objID); } + // cacheImpl -> cacheObject inserts gaps for remaining objects and groups constexpr auto kEndOfTrack = ObjectStatus::END_OF_TRACK; auto res = cacheImpl( gID, sgID, objID, kEndOfTrack, noExtensions(), nullptr, true, true); @@ -1020,7 +1160,8 @@ class MoQCache::FetchWriteback : public FetchConsumer { } folly::Expected endOfFetch() override { - cacheMissing(fetchRangeIt_.end()); + // Mark all remaining positions as known (upstream has completed) + markNonExistentTo(fetchRangeIt_.end()); updateInProgress(); complete_.post(); if (proxyFin_) { @@ -1098,24 +1239,25 @@ class MoQCache::FetchWriteback : public FetchConsumer { MoQCache& cache_; FullTrackName ftn_; - void cacheMissing(AbsoluteLocation current) { - while (*fetchRangeIt_ != current) { - auto& group = fetchRangeIt_.track->getOrCreateGroup(fetchRangeIt_->group); - if (fetchRangeIt_->group != current.group) { - if (fetchRangeIt_->object == 0) { - fetchRangeIt_.track->updateLargest({fetchRangeIt_->group, 0}); - group.cacheMissingStatus(0, ObjectStatus::GROUP_NOT_EXIST); - } else { - group.endOfGroup = true; - } - } else { - fetchRangeIt_.track->updateLargest( - {fetchRangeIt_->group, fetchRangeIt_->object}); - group.cacheMissingStatus( - fetchRangeIt_->object, ObjectStatus::OBJECT_NOT_EXIST); - } - fetchRangeIt_.next(); + void markNonExistentTo(AbsoluteLocation target) { + // Mark all positions from current iterator position up to (but not + // including) target as nonexistent, using range inserts for efficiency. + if (*fetchRangeIt_ == target) { + return; + } + + auto ranges = getGapRanges( + *fetchRangeIt_, + target, + fetchRangeIt_.order, + fetchRangeIt_.minLocation, + fetchRangeIt_.maxLocation); + for (const auto& [start, end] : ranges) { + fetchRangeIt_.track->gaps.insert(start, end); } + + // Advance iterator directly to target + fetchRangeIt_.advanceTo(target); } folly::Expected cacheImpl( @@ -1131,6 +1273,8 @@ class MoQCache::FetchWriteback : public FetchConsumer { auto& group = fetchRangeIt_.track->getOrCreateGroup(groupID); auto cacheRes = cache_.cacheObjectAndUpdateBytes( group, + *fetchRangeIt_.track, + groupID, subgroupID, objectID, status, @@ -1139,7 +1283,9 @@ class MoQCache::FetchWriteback : public FetchConsumer { complete, forwardingPreferenceIsDatagram, cache_.now()); - cacheMissing({groupID, objectID}); + // In a fetch, positions between received objects are known to not exist + // (fetch is ordered, gaps indicate non-existence) + markNonExistentTo({groupID, objectID}); if (cacheRes.hasError()) { updateInProgress(); return cacheRes; @@ -1154,7 +1300,10 @@ class MoQCache::FetchWriteback : public FetchConsumer { fetchRangeIt_.next(); updateInProgress(); if (finFetch) { - cacheMissing(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(); } } @@ -1219,7 +1368,7 @@ folly::coro::Task MoQCache::fetch( last.object--; } else { // if end is 1,0, that means all of group 1 - last.object = std::numeric_limits::max(); + last.object = kLocationMax.object; standalone->end.group++; // TODO: handle case where track.largestGroupAndObject is an END_OF_GROUP // or END_OF_TRACK @@ -1303,11 +1452,16 @@ folly::coro::Task MoQCache::fetchImpl( << current.object << "}"; co_await writeback->waitFor(current); cachedNow = now(); + // The in-progress fetch may have inserted new gap info + fetchRangeIt.skipGaps(); + if (*fetchRangeIt != current) { + // Gap was inserted at this position; restart loop + continue; + } } - // Gets cached object if cache hit, else none. - auto cachedObjectMaybe = getCachedObjectMaybe(*track, current, cachedNow); - if (!cachedObjectMaybe) { + auto* cachedObject = getCachedObjectMaybe(*track, current, cachedNow); + if (!cachedObject) { if (!fetchStart) { fetchStart = current; } @@ -1315,7 +1469,6 @@ folly::coro::Task MoQCache::fetchImpl( continue; } - // found the object, first fetch missing range, if any XLOG(DBG1) << "object cache HIT for {" << current.group << "," << current.object << "}"; // NOTE: This raw CacheEntry* is safe despite the fetchUpstream call below @@ -1351,7 +1504,7 @@ folly::coro::Task MoQCache::fetchImpl( fetchStart.reset(); } XLOG(DBG1) << "Publish object from cache"; - auto object = cachedObjectMaybe.value(); + auto* object = cachedObject; fetchRangeIt.next(); lastObject = (*fetchRangeIt) == fetchRangeIt.end(); auto res = @@ -1424,9 +1577,12 @@ folly::coro::Task MoQCache::fetchImpl( co_return nullptr; } } + // Call endOfFetch if we didn't serve an object with fin=true + if (!(lastObject && servedOneObject)) { + consumer->endOfFetch(); + } if (!fetchHandle) { XLOG(DBG1) << "Fetch completed entirely from cache"; - // test for empty range with no largest group and object? if (servedOneObject) { if (standalone->end.object == 0) { standalone->end.group--; @@ -1440,7 +1596,6 @@ folly::coro::Task MoQCache::fetchImpl( co_return std::make_shared(FetchOk{ fetch.requestID, fetch.groupOrder, endOfTrack, standalone->end, {}}); } else { - consumer->endOfFetch(); co_return std::make_shared(FetchOk{ fetch.requestID, fetch.groupOrder, false, standalone->end, {}}); } @@ -1448,26 +1603,26 @@ folly::coro::Task MoQCache::fetchImpl( co_return nullptr; } -std::optional MoQCache::getCachedObjectMaybe( +// Returns valid CacheEntry* on cache hit, nullptr on miss. +// Gap-skipping is handled by FetchRangeIterator before this is called. +MoQCache::CacheEntry* MoQCache::getCachedObjectMaybe( CacheTrack& track, AbsoluteLocation current, TimePoint now) { auto groupIt = track.groups.find(current.group); - // Group missing from cache, advance. if (groupIt == track.groups.end()) { - // object not cached or incomplete, count as miss. XLOG(DBG1) << "group cache miss for {" << current.group << "}"; - return std::nullopt; + return nullptr; } auto& group = groupIt->second; auto objIt = group->objects.find(current.object); - if (objIt == group->objects.end() || !objIt->second->complete) { - // object not cached or incomplete, count as miss. + if (objIt == group->objects.end()) { XLOG(DBG1) << "object cache miss for {" << current.group << "," << current.object << "}"; - return std::nullopt; + return nullptr; } + auto effectiveDuration = track.maxCacheDuration ? track.maxCacheDuration : defaultMaxCacheDuration_; if (effectiveDuration) { @@ -1479,10 +1634,17 @@ std::optional MoQCache::getCachedObjectMaybe( group->totalBytes -= expiredBytes; totalCachedBytes_ -= expiredBytes; group->objects.erase(objIt); - return std::nullopt; + return nullptr; } } - return std::make_optional(objIt->second.get()); + + if (!objIt->second->complete) { + XLOG(DBG1) << "object incomplete for {" << current.group << "," + << current.object << "}"; + return nullptr; + } + + return objIt->second.get(); } folly::coro::Task MoQCache::fetchUpstream( @@ -1626,47 +1788,45 @@ MoQCache::FetchRangeIterator::FetchRangeIterator( // But objects are iterarted in ascending order // Same group, objs ascending. - if (start.group == end.group) { - return; - } - - // --- set iterator start --- - // If the last group object == 0, current needs to be - // one group down, and the object starts at: - // 0 if its not the start group - // start.object if it is - if (end_.object == 0) { - current_.group = end_.group - 1; - if (current_.group == start.group) { - current_.object = start.object; + if (start.group != end.group) { + // --- set iterator start --- + // If the last group object == 0, current needs to be + // one group down, and the object starts at: + // 0 if its not the start group + // start.object if it is + if (end_.object == 0) { + current_.group = end_.group - 1; + if (current_.group == start.group) { + current_.object = start.object; + } else { + current_.object = 0; + } } else { + // Set current to first obj of last group + current_ = end_; current_.object = 0; } - } else { - // Set current to first obj of last group - current_ = end_; - current_.object = 0; - } - // --- set iterator end --- - // Set end to the last object of the first group + 1 - // This could be endOfGroup if known, or max_int - auto groupIt = track->groups.find(start.group); - end_ = start; - if (groupIt == track->groups.end()) { - // No group found, so we set it to max for now - // Track update at callback will signal endOfGroup - end_.object = std::numeric_limits::max(); - return; - } - - auto& group = groupIt->second; - end_.object = std::numeric_limits::max(); - if (group->endOfGroup && - group->maxCachedObject < std::numeric_limits::max()) { - end_.object = group->maxCachedObject + 1; + // --- set iterator end --- + // Set end to the last object of the first group + 1 + // This could be endOfGroup if known, or max_int + auto groupIt = track->groups.find(start.group); + end_ = start; + if (groupIt != track->groups.end()) { + auto& group = groupIt->second; + end_.object = kLocationMax.object; + if (group->endOfGroup && group->maxCachedObject < kLocationMax.object) { + end_.object = group->maxCachedObject + 1; + } + } else { + // No group found, so we set it to max for now + // Track update at callback will signal endOfGroup + end_.object = kLocationMax.object; + } } } + // Skip any gaps at initial position + skipGaps(); } void MoQCache::FetchRangeIterator::invalidate() { @@ -1674,6 +1834,28 @@ void MoQCache::FetchRangeIterator::invalidate() { isValid_ = false; } +void MoQCache::FetchRangeIterator::advanceTo(const AbsoluteLocation& loc) { + current_ = loc; + // Check if we've passed the valid range based on iteration direction + // Note: use strict inequality - the loop condition handles reaching exactly + // the end + if (order != GroupOrder::NewestFirst) { + // Ascending: invalid if current > maxLocation (passed the end) + if (current_ > maxLocation) { + isValid_ = false; + return; + } + } else { + // Descending: invalid if current < minLocation (passed the minimum) + if (current_ < minLocation) { + isValid_ = false; + return; + } + } + // Skip any gaps at new position + skipGaps(); +} + const AbsoluteLocation& MoQCache::FetchRangeIterator::operator*() const { return current_; } @@ -1683,16 +1865,27 @@ const AbsoluteLocation* MoQCache::FetchRangeIterator::operator->() const { } void MoQCache::FetchRangeIterator::next() { + if (!isValid_) { + return; + } + advanceOne(); + skipGaps(); +} + +void MoQCache::FetchRangeIterator::advanceOne() { if (!isValid_) { return; } if (current_ == end_) { - isValid_ = false; + XLOG(DBG2) << "advanceOne: current_ == end_, staying at end"; return; } auto groupEndMaybe = findGroupEndMaybe(current_.group, cachedGroupId_, cachedGroupPtr_); + XLOG(DBG2) << "advanceOne: current={" << current_.group << "," + << current_.object << "} groupEndMaybe=" + << (groupEndMaybe ? std::to_string(*groupEndMaybe) : "nullopt"); if (groupEndMaybe && current_.object < groupEndMaybe.value()) { // Found group end, we can increment object till that current_.object++; @@ -1700,8 +1893,7 @@ void MoQCache::FetchRangeIterator::next() { if (order == GroupOrder::NewestFirst) { // desc if (current_.group == 0) { - XLOG(ERR) << "GroupID underflow in FetchRangeIter: Current groupID " - << current_.group << " CurrentObjID=" << current_.object; + XLOG(DBG2) << "advanceOne: at group 0, invalidating iterator"; isValid_ = false; return; } @@ -1713,16 +1905,68 @@ void MoQCache::FetchRangeIterator::next() { } else { // asc if (current_.group == maxLocation.group) { - XLOG(ERR) - << "GroupID beyond maxLocation in FetchRangeIter: Current groupID " - << current_.group << " CurrentObjID=" << current_.object; - isValid_ = false; + XLOG(DBG2) << "advanceOne: at maxLocation.group, clamping to end"; + current_ = end_; return; } current_.group++; current_.object = 0; } } + XLOG(DBG2) << "advanceOne: new current={" << current_.group << "," + << current_.object << "}"; +} + +void MoQCache::FetchRangeIterator::skipGaps() { + while (isValid_ && track->gaps.contains(current_)) { + XLOG(DBG2) << "skipGaps: current={" << current_.group << "," + << current_.object << "} is in gap"; + auto gapInterval = track->gaps.findInterval(current_); + if (!gapInterval) { + // Shouldn't happen, but fallback to single step + advanceOne(); + continue; + } + + auto [gapStart, gapEnd] = *gapInterval; + XLOG(DBG2) << "skipGaps: gap=[{" << gapStart.group << "," << gapStart.object + << "}, {" << gapEnd.group << "," << gapEnd.object << "}]"; + std::optional afterGap; + + if (order != GroupOrder::NewestFirst) { + // Ascending: skip to position after gap end + afterGap = gapEnd.next(); + } else { + afterGap = computeDescendingAfterGap(current_, gapStart, minLocation); + } + + if (afterGap) { + XLOG(DBG2) << "skipGaps: afterGap={" << afterGap->group << "," + << afterGap->object << "}"; + // Check bounds and clamp to end if needed + // For ascending: afterGap must be < end_ (before exclusive end) + // For descending: afterGap must be >= minLocation (within fetch range) + bool inRange = (order != GroupOrder::NewestFirst) + ? (*afterGap < end_) + : (*afterGap >= minLocation); + if (inRange) { + current_ = *afterGap; + XLOG(DBG2) << "skipGaps: moved to {" << current_.group << "," + << current_.object << "}"; + } else { + // Gap extends to or past end - clamp to end + XLOG(DBG2) << "skipGaps: clamping to end {" << end_.group << "," + << end_.object << "}"; + current_ = end_; + return; // No point continuing the loop + } + } else { + // Gap extends to boundary - clamp to end + XLOG(DBG2) << "skipGaps: no afterGap, clamping to end"; + current_ = end_; + return; + } + } } /* @@ -1767,13 +2011,11 @@ std::optional MoQCache::FetchRangeIterator::findGroupEndMaybe( // Calculate object boundaries (with overflow protection) const uint64_t endIncludingMaxCached = - (maxObjectInGroup < std::numeric_limits::max()) - ? maxObjectInGroup + 1 - : maxObjectInGroup; + (maxObjectInGroup < kLocationMax.object) ? maxObjectInGroup + 1 + : maxObjectInGroup; const uint64_t firstUncachedObject = - (endIncludingMaxCached < std::numeric_limits::max()) - ? endIncludingMaxCached + 1 - : endIncludingMaxCached; + (endIncludingMaxCached < kLocationMax.object) ? endIncludingMaxCached + 1 + : endIncludingMaxCached; // Handle first group if (isFirstGroup) { @@ -2027,6 +2269,8 @@ bool MoQCache::evictForByteLimitIfNeeded() { folly::Expected MoQCache::cacheObjectAndUpdateBytes( CacheGroup& group, + CacheTrack& track, + uint64_t groupID, uint64_t subgroup, uint64_t objectID, ObjectStatus status, @@ -2037,6 +2281,8 @@ MoQCache::cacheObjectAndUpdateBytes( TimePoint now) { size_t oldGroupBytes = group.totalBytes; auto res = group.cacheObject( + track, + groupID, subgroup, objectID, status, diff --git a/moxygen/relay/MoQCache.h b/moxygen/relay/MoQCache.h index 0cb76fedc..89f37e75a 100644 --- a/moxygen/relay/MoQCache.h +++ b/moxygen/relay/MoQCache.h @@ -14,9 +14,9 @@ #include #include #include +#include #include #include -#include namespace moxygen { @@ -82,7 +82,7 @@ class MoQCache { if (it == cache_.end()) { return false; } - return getCachedObjectMaybe(*it->second, obj, now()).has_value(); + return getCachedObjectMaybe(*it->second, obj, now()) != nullptr; } // Setters for testing - update cache limits and evict if necessary @@ -176,10 +176,14 @@ class MoQCache { class SubgroupWriteback; class FetchWriteback; class FetchHandle; + struct CacheTrack; // Forward declaration for CacheGroup::cacheObject // Entry for a group struct CacheGroup { folly::F14FastMap> objects; + // TODO: may be redundant now that FetchRangeIterator::skipGaps() uses gap + // data to advance past group boundaries. Consider removing along with + // findGroupEndMaybe. uint64_t maxCachedObject{0}; bool endOfGroup{false}; // Track seen Prior Group ID Gap value for validation @@ -191,6 +195,8 @@ class MoQCache { folly::Optional::iterator> lruIter_; folly::Expected cacheObject( + CacheTrack& track, + uint64_t groupID, uint64_t subgroup, uint64_t objectID, ObjectStatus status, @@ -199,7 +205,6 @@ class MoQCache { bool complete, bool forwardingPreferenceIsDatagram, TimePoint now); - void cacheMissingStatus(uint64_t objectID, ObjectStatus status); }; // Entry for a track @@ -210,6 +215,7 @@ class MoQCache { struct CacheTrack { folly::F14FastMap> groups; + LocationIntervalSet gaps; size_t liveWritebackCount{0}; bool endOfTrack{false}; std::optional largestGroupAndObject; @@ -234,7 +240,7 @@ class MoQCache { CacheGroup& getOrCreateGroup(uint64_t groupID, MoQCache* cache = nullptr); // Process Prior Group ID Gap and Prior Object ID Gap extensions - // and cache the missing groups/objects accordingly + // and mark the NonExistent groups/objects accordingly folly::Expected processGapExtensions( uint64_t groupID, uint64_t objectID, @@ -247,7 +253,7 @@ class MoQCache { }; // Group-order-aware iterator for traversing groups (objects within groups - // always ascending) + // always ascending). Automatically skips over gaps. class FetchRangeIterator { public: FetchRangeIterator( @@ -258,6 +264,8 @@ class MoQCache { AbsoluteLocation end(); void next(); + void skipGaps(); // Skip over any gaps at current position + void advanceTo(const AbsoluteLocation& loc); const AbsoluteLocation& operator*() const; const AbsoluteLocation* operator->() const; void invalidate(); @@ -269,12 +277,13 @@ class MoQCache { std::shared_ptr track; private: + void advanceOne(); // Single step without gap skipping AbsoluteLocation current_; AbsoluteLocation end_; bool isValid_ = true; - mutable uint64_t cachedGroupId_{std::numeric_limits::max()}; + mutable uint64_t cachedGroupId_{kLocationMax.group}; mutable std::shared_ptr cachedGroupPtr_{nullptr}; - mutable uint64_t cachedEndGroupId_{std::numeric_limits::max()}; + mutable uint64_t cachedEndGroupId_{kLocationMax.group}; mutable std::shared_ptr cachedEndGroupPtr_{nullptr}; std::optional findGroupEndMaybe( uint64_t groupId, @@ -305,7 +314,10 @@ class MoQCache { // Injectable clock for testing std::function clock_; - std::optional + // Returns: + // Returns valid CacheEntry* on cache hit, nullptr on miss. + // Caller (fetchImpl) is responsible for gap-skipping via FetchRangeIterator. + CacheEntry* getCachedObjectMaybe(CacheTrack& track, AbsoluteLocation obj, TimePoint now); folly::coro::Task fetchImpl( @@ -349,6 +361,8 @@ class MoQCache { // Wraps cacheObject() + byte accounting + eviction check folly::Expected cacheObjectAndUpdateBytes( CacheGroup& group, + CacheTrack& track, + uint64_t groupID, uint64_t subgroup, uint64_t objectID, ObjectStatus status, diff --git a/moxygen/relay/test/MoQCacheTests.cpp b/moxygen/relay/test/MoQCacheTests.cpp index cd48e6679..3919afc98 100644 --- a/moxygen/relay/test/MoQCacheTests.cpp +++ b/moxygen/relay/test/MoQCacheTests.cpp @@ -553,7 +553,7 @@ CO_TEST_F(MoQCacheTest, TestFetchPopulatesNotExist) { co_await folly::coro::co_reschedule_on_current_executor; - expectFetchObjects({0, 0}, {2, 10}, true, 10, 2, 2); + expectFetchObjects({0, 0}, {2, 10}, false, 10, 2, 2); res = co_await cache_.fetch(getFetch({0, 0}, {2, 10}), consumer_, upstream_); EXPECT_TRUE(res.hasValue()); co_return; @@ -825,9 +825,9 @@ CO_TEST_F(MoQCacheTest, TestFetchPopulatesNotExistObjectsAndGroups) { writeback->objectStream(header2, nullptr); writeback.reset(); - EXPECT_CALL(*consumer_, endOfGroup(0, 0, 1, false)) + // With gap skipping in next(), the last object gets fin=true + EXPECT_CALL(*consumer_, endOfGroup(0, 0, 1, true)) .WillOnce(Return(folly::unit)); - EXPECT_CALL(*consumer_, endOfFetch()).WillOnce(Return(folly::unit)); auto res = co_await cache_.fetch(getFetch({0, 0}, {1, 0}), consumer_, upstream_); EXPECT_TRUE(res.hasValue()); @@ -842,9 +842,9 @@ TEST_F(MoQCacheTest, TestInvalidCacheUpdateFails) { ObjectHeader header5(5, 0, 0, 0, 10); header5.extensions = makeGroupGapExtensions(5); writeback->objectStream(header5, makeBuf(10)); - + // leave an unknown gap at 6, we use it later to test other APIs writeback->objectStream( - ObjectHeader(6, 0, 0, 0, ObjectStatus::END_OF_TRACK), nullptr); + ObjectHeader(7, 0, 0, 0, ObjectStatus::END_OF_TRACK), nullptr); writeback.reset(); @@ -855,32 +855,33 @@ TEST_F(MoQCacheTest, TestInvalidCacheUpdateFails) { auto result = writeback->objectStream(ObjectHeader(0, 0, 0, 0, 100), makeBuf(100)); EXPECT_TRUE(result.hasError()); - EXPECT_EQ(result.error().code, MoQPublishError::API_ERROR); + EXPECT_EQ(result.error().code, MoQPublishError::MALFORMED_TRACK); result = writeback->datagram(ObjectHeader(1, 0, 0, 0, 100), makeBuf(100)); EXPECT_TRUE(result.hasError()); - EXPECT_EQ(result.error().code, MoQPublishError::API_ERROR); + EXPECT_EQ(result.error().code, MoQPublishError::MALFORMED_TRACK); - result = writeback->beginSubgroup(2, 0, 0).value()->endOfGroup(0); + result = writeback->objectStream( + ObjectHeader(2, 0, 0, 0, ObjectStatus::END_OF_GROUP), nullptr); EXPECT_TRUE(result.hasError()); - EXPECT_EQ(result.error().code, MoQPublishError::API_ERROR); + EXPECT_EQ(result.error().code, MoQPublishError::MALFORMED_TRACK); - result = writeback->beginSubgroup(3, 0, 0).value()->beginObject( - 0, 100, makeBuf(100)); + result = writeback->objectStream(ObjectHeader(3, 0, 0, 0, 100), makeBuf(100)); EXPECT_TRUE(result.hasError()); - EXPECT_EQ(result.error().code, MoQPublishError::API_ERROR); + EXPECT_EQ(result.error().code, MoQPublishError::MALFORMED_TRACK); - result = writeback->beginSubgroup(4, 0, 0).value()->endOfTrackAndGroup(0); + result = writeback->objectStream( + ObjectHeader(4, 0, 0, 0, ObjectStatus::END_OF_TRACK), nullptr); EXPECT_TRUE(result.hasError()); EXPECT_EQ(result.error().code, MoQPublishError::MALFORMED_TRACK); // Payload size changed result = writeback->objectStream(ObjectHeader(5, 0, 0, 0, 20), makeBuf(20)); EXPECT_TRUE(result.hasError()); - EXPECT_EQ(result.error().code, MoQPublishError::API_ERROR); + EXPECT_EQ(result.error().code, MoQPublishError::MALFORMED_TRACK); // Beyond End of track - result = writeback->objectStream(ObjectHeader(7, 0, 0, 0, 20), makeBuf(20)); + result = writeback->objectStream(ObjectHeader(8, 0, 0, 0, 20), makeBuf(20)); EXPECT_TRUE(result.hasError()); EXPECT_EQ(result.error().code, MoQPublishError::MALFORMED_TRACK); @@ -895,7 +896,7 @@ TEST_F(MoQCacheTest, TestInvalidCacheUpdateFails) { EXPECT_FALSE(result.hasError()); writeback->beginSubgroup(6, 0, 0).value()->checkpoint(); - writeback->beginSubgroup(7, 0, 0).value()->reset( + writeback->beginSubgroup(6, 0, 0).value()->reset( ResetStreamErrorCode::CANCELLED); writeback->publishDone( {RequestID(0), PublishDoneStatusCode::SUBSCRIPTION_ENDED, 0, ""}); @@ -1077,6 +1078,42 @@ CO_TEST_F(MoQCacheTest, TestUpstreamFetchUsingBeginObjectAndObjectPayload) { EXPECT_EQ(res.value()->fetchOk().endLocation, (AbsoluteLocation{0, 0})); } +CO_TEST_F(MoQCacheTest, TestObjectPayloadMarksRemainingAsNonexistent) { + // Fetch for [0,0] - [0,2] exclusive (objects 0 and 1) + // Upstream returns only object 0 via beginObject + objectPayload with + // finFetch=true Object 1 should be marked as nonexistent + expectUpstreamFetch({0, 0}, {0, 2}, 0, AbsoluteLocation{0, 0}) + .via(co_await folly::coro::co_current_executor) + .thenTry([this](const auto&) { + // Send only object 0, then finish the fetch + upstreamFetchConsumer_->beginObject(0, 0, 0, 100, makeBuf(50)); + auto status = upstreamFetchConsumer_->objectPayload( + makeBuf(50), /*finSubgroup=*/true); + EXPECT_FALSE(status.hasError()); + }); + + // Expect the consumer to receive object 0 + EXPECT_CALL(*consumer_, beginObject(0, 0, 0, 100, _, _)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*consumer_, objectPayload(_, true)) + .WillOnce(Return(ObjectPublishStatus::DONE)); + + // Perform the fetch + auto res = + co_await cache_.fetch(getFetch({0, 0}, {0, 2}), consumer_, upstream_); + EXPECT_TRUE(res.hasValue()); + EXPECT_EQ(res.value()->fetchOk().endLocation, (AbsoluteLocation{0, 0})); + + // Subsequent fetch for [0,1] - [0,2] should return cached non-existence + // (no upstream call, just endOfFetch) + EXPECT_CALL(*consumer_, endOfFetch()).WillOnce(Return(folly::unit)); + + auto res2 = + co_await cache_.fetch(getFetch({0, 1}, {0, 2}), consumer_, upstream_); + EXPECT_TRUE(res2.hasValue()); + EXPECT_EQ(res2.value()->fetchOk().endLocation, (AbsoluteLocation{0, 2})); +} + CO_TEST_F(MoQCacheTest, TestPopulateCacheWithBeginSubgroupAndFetch) { // Create a writeback for the test track auto writeback = cache_.getSubscribeWriteback(kTestTrackName, trackConsumer_); @@ -1733,7 +1770,463 @@ TEST_F(MoQCacheTest, TestPriorObjectIdGapOverlappingNotExistOnly) { auto result2 = writeback->datagram(header2, makeBuf(100)); // Currently fails - see open spec issue about out-of-order datagram delivery EXPECT_TRUE(result2.hasError()); - EXPECT_EQ(result2.error().code, MoQPublishError::API_ERROR); + EXPECT_EQ(result2.error().code, MoQPublishError::MALFORMED_TRACK); +} + +CO_TEST_F(MoQCacheTest, TestLargeGroupGap) { + // Test that caching/fetching with a large group gap (2^60) is O(1), + // not O(groups). This verifies IntervalSet range operations are used. + constexpr uint64_t kLargeGroup = 1ULL << 60; + + // Cache object at {0, 0} + populateCacheRange({0, 0}, {0, 1}); + + // Cache object at {kLargeGroup, 0} with group gap extension + // Gap value is kLargeGroup - 1 (groups 1 through kLargeGroup-1 don't exist) + // This should mark groups 1 through kLargeGroup-1 as nonexistent in O(1) + auto writeback = cache_.getSubscribeWriteback(kTestTrackName, trackConsumer_); + ObjectHeader header(kLargeGroup, 0, 0, 0, 100); + header.extensions = makeGroupGapExtensions(kLargeGroup - 1); + auto result = writeback->datagram(header, makeBuf(100)); + EXPECT_TRUE(result.hasValue()); + writeback.reset(); + + // Fetch {0, 0} - should be a cache hit + expectFetchObjects({0, 0}, {0, 1}, false); + auto res = + co_await cache_.fetch(getFetch({0, 0}, {0, 1}), consumer_, upstream_); + EXPECT_TRUE(res.hasValue()); + + // Fetch a group in the middle of the gap - should return immediately + // as known nonexistent (no upstream fetch needed) + auto midGapConsumer = std::make_shared>(); + EXPECT_CALL(*midGapConsumer, endOfFetch()).WillOnce(Return(folly::unit)); + res = co_await cache_.fetch( + getFetch({kLargeGroup / 2, 0}, {kLargeGroup / 2, 1}), + midGapConsumer, + upstream_); + EXPECT_TRUE(res.hasValue()); + + // Fetch {kLargeGroup, 0} - should be a cache hit + auto largeGroupConsumer = std::make_shared>(); + EXPECT_CALL( + *largeGroupConsumer, + object(kLargeGroup, 0, 0, HasChainDataLengthOf(100), _, true, true)) + .WillOnce(Return(folly::unit)); + // No endOfFetch() expected - object was delivered with fin=true + res = co_await cache_.fetch( + getFetch({kLargeGroup, 0}, {kLargeGroup, 1}), + largeGroupConsumer, + upstream_); + EXPECT_TRUE(res.hasValue()); + + // Test descending order fetch in the gap - should skip efficiently + auto descGapConsumer = std::make_shared>(); + EXPECT_CALL(*descGapConsumer, endOfFetch()).WillOnce(Return(folly::unit)); + res = co_await cache_.fetch( + getFetch( + {kLargeGroup / 2, 0}, {kLargeGroup / 2, 1}, GroupOrder::NewestFirst), + descGapConsumer, + upstream_); + EXPECT_TRUE(res.hasValue()); +} + +CO_TEST_F(MoQCacheTest, TestDescendingFetchGapSkipping) { + // Test that descending fetch correctly skips gaps in O(1) + // Scenario: gaps at various locations, fetch in descending order + + // Cache objects at groups 0, 5, and 10 (with END_OF_GROUP to mark end) + auto writeback = cache_.getSubscribeWriteback(kTestTrackName, trackConsumer_); + + // Group 0 with object 0 as END_OF_GROUP + ObjectHeader header0(0, 0, 0, 0, ObjectStatus::END_OF_GROUP); + writeback->objectStream(header0, makeBuf(100)); + + // Group 5 with object 0 and gap extension for groups 1-4 + ObjectHeader header5(5, 0, 0, 0, ObjectStatus::END_OF_GROUP); + header5.extensions = makeGroupGapExtensions(4); // Groups 1-4 don't exist + writeback->objectStream(header5, makeBuf(100)); + + // Group 10 with object 0 and gap extension for groups 6-9 + ObjectHeader header10(10, 0, 0, 0, ObjectStatus::END_OF_GROUP); + header10.extensions = makeGroupGapExtensions(4); // Groups 6-9 don't exist + writeback->objectStream(header10, makeBuf(100)); + writeback.reset(); + + // Descending fetch from {10, 0} to {0, 0} - should hit 10, skip 9-6, hit 5, + // skip 4-1, hit 0 + auto descConsumer = std::make_shared>(); + { + InSequence seq; + EXPECT_CALL(*descConsumer, endOfGroup(10, 0, 0, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*descConsumer, endOfGroup(5, 0, 0, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*descConsumer, endOfGroup(0, 0, 0, true)) + .WillOnce(Return(folly::unit)); + } + auto res = co_await cache_.fetch( + getFetch({0, 0}, {10, 1}, GroupOrder::NewestFirst), + descConsumer, + upstream_); + EXPECT_TRUE(res.hasValue()); +} + +CO_TEST_F(MoQCacheTest, TestGapSpanningGroups) { + // Test gap that spans multiple groups (both ascending and descending) + + auto writeback = cache_.getSubscribeWriteback(kTestTrackName, trackConsumer_); + + // Cache object at {0, 0} as END_OF_GROUP + ObjectHeader header0(0, 0, 0, 0, ObjectStatus::END_OF_GROUP); + writeback->objectStream(header0, makeBuf(100)); + + // Cache object at {100, 0} with gap extension marking groups 1-99 + ObjectHeader header100(100, 0, 0, 0, ObjectStatus::END_OF_GROUP); + header100.extensions = makeGroupGapExtensions(99); // Groups 1-99 don't exist + writeback->objectStream(header100, makeBuf(100)); + writeback.reset(); + + // Ascending fetch spanning gap - should hit 0, skip 1-99, hit 100 + auto ascConsumer = std::make_shared>(); + { + InSequence seq; + EXPECT_CALL(*ascConsumer, endOfGroup(0, 0, 0, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*ascConsumer, endOfGroup(100, 0, 0, true)) + .WillOnce(Return(folly::unit)); + } + auto res = + co_await cache_.fetch(getFetch({0, 0}, {100, 1}), ascConsumer, upstream_); + EXPECT_TRUE(res.hasValue()); + + // Descending fetch spanning gap - should hit 100, skip 99-1, hit 0 + auto descConsumer = std::make_shared>(); + { + InSequence seq; + EXPECT_CALL(*descConsumer, endOfGroup(100, 0, 0, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*descConsumer, endOfGroup(0, 0, 0, true)) + .WillOnce(Return(folly::unit)); + } + res = co_await cache_.fetch( + getFetch({0, 0}, {100, 1}, GroupOrder::NewestFirst), + descConsumer, + upstream_); + EXPECT_TRUE(res.hasValue()); +} + +CO_TEST_F(MoQCacheTest, TestInGroupObjectGapSkipping) { + // Test gap within a single group (object-level gaps) + // The fetch range must exactly cover the known objects to avoid cache misses + + auto writeback = cache_.getSubscribeWriteback(kTestTrackName, trackConsumer_); + + // Object 0 - normal object (group=0, subgroup=0, id=0) + ObjectHeader header0(0, 0, 0, 0, ObjectStatus::NORMAL); + writeback->objectStream(header0, makeBuf(100)); + + // Object 50 with gap extension marking objects 1-49 (group=0, subgroup=0, + // id=50) + ObjectHeader header50(0, 0, 50, 0, ObjectStatus::NORMAL); + header50.extensions = makeObjectGapExtensions(49); + writeback->objectStream(header50, makeBuf(100)); + + // Object 100 as END_OF_GROUP with gap extension marking objects 51-99 + ObjectHeader header100(0, 0, 100, 0, ObjectStatus::END_OF_GROUP); + header100.extensions = makeObjectGapExtensions(49); + writeback->objectStream(header100, nullptr); + writeback.reset(); + + // Ascending fetch within the group + auto ascConsumer = std::make_shared>(); + { + InSequence seq; + EXPECT_CALL( + *ascConsumer, + object(0, 0, 0, HasChainDataLengthOf(100), _, false, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL( + *ascConsumer, + object(0, 0, 50, HasChainDataLengthOf(100), _, false, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*ascConsumer, endOfGroup(0, 0, 100, true)) + .WillOnce(Return(folly::unit)); + } + // Fetch from {0, 0} to {0, 101} to cover all objects in group 0 + auto res = + co_await cache_.fetch(getFetch({0, 0}, {0, 101}), ascConsumer, upstream_); + EXPECT_TRUE(res.hasValue()); +} + +CO_TEST_F(MoQCacheTest, TestDescendingSingleIntermediateGroup) { + // Validates the off-by-one fix in getGapRanges: when there is exactly one + // intermediate group between start and end in descending order, it must be + // marked as a gap. + + auto writeback = cache_.getSubscribeWriteback(kTestTrackName, trackConsumer_); + + // Group 3 with object 0 as END_OF_GROUP + ObjectHeader header3(3, 0, 0, 0, ObjectStatus::END_OF_GROUP); + writeback->objectStream(header3, makeBuf(100)); + + // Group 5 with object 0 as END_OF_GROUP, gap extension for group 4 + ObjectHeader header5(5, 0, 0, 0, ObjectStatus::END_OF_GROUP); + header5.extensions = makeGroupGapExtensions(1); // Group 4 doesn't exist + writeback->objectStream(header5, makeBuf(100)); + writeback.reset(); + + // Descending fetch from {5, 0} to {3, 0} - should hit 5, skip 4, hit 3 + auto descConsumer = std::make_shared>(); + { + InSequence seq; + EXPECT_CALL(*descConsumer, endOfGroup(5, 0, 0, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*descConsumer, endOfGroup(3, 0, 0, true)) + .WillOnce(Return(folly::unit)); + } + auto res = co_await cache_.fetch( + getFetch({3, 0}, {5, 1}, GroupOrder::NewestFirst), + descConsumer, + upstream_); + EXPECT_TRUE(res.hasValue()); +} + +CO_TEST_F(MoQCacheTest, TestDescendingAdjacentGroups) { + // Adjacent groups with no intermediate group — verify no gap issues. + + auto writeback = cache_.getSubscribeWriteback(kTestTrackName, trackConsumer_); + + ObjectHeader header3(3, 0, 0, 0, ObjectStatus::END_OF_GROUP); + writeback->objectStream(header3, makeBuf(100)); + + ObjectHeader header4(4, 0, 0, 0, ObjectStatus::END_OF_GROUP); + writeback->objectStream(header4, makeBuf(100)); + writeback.reset(); + + auto descConsumer = std::make_shared>(); + { + InSequence seq; + EXPECT_CALL(*descConsumer, endOfGroup(4, 0, 0, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*descConsumer, endOfGroup(3, 0, 0, true)) + .WillOnce(Return(folly::unit)); + } + auto res = co_await cache_.fetch( + getFetch({3, 0}, {4, 1}, GroupOrder::NewestFirst), + descConsumer, + upstream_); + EXPECT_TRUE(res.hasValue()); +} + +CO_TEST_F(MoQCacheTest, TestDescendingGapNearGroupZero) { + // Groups 0 and 2 cached, group 1 is a gap, descending fetch. + + auto writeback = cache_.getSubscribeWriteback(kTestTrackName, trackConsumer_); + + ObjectHeader header0(0, 0, 0, 0, ObjectStatus::END_OF_GROUP); + writeback->objectStream(header0, makeBuf(100)); + + ObjectHeader header2(2, 0, 0, 0, ObjectStatus::END_OF_GROUP); + header2.extensions = makeGroupGapExtensions(1); // Group 1 doesn't exist + writeback->objectStream(header2, makeBuf(100)); + writeback.reset(); + + auto descConsumer = std::make_shared>(); + { + InSequence seq; + EXPECT_CALL(*descConsumer, endOfGroup(2, 0, 0, false)) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(*descConsumer, endOfGroup(0, 0, 0, true)) + .WillOnce(Return(folly::unit)); + } + auto res = co_await cache_.fetch( + getFetch({0, 0}, {2, 1}, GroupOrder::NewestFirst), + descConsumer, + upstream_); + 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>(); + 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 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(FetchOk{ + 0, + GroupOrder::NewestFirst, + false, + AbsoluteLocation{3, 3}, + }); + return folly::coro::makeTask( + 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>(); + ON_CALL(*laterConsumer, endOfFetch()).WillByDefault(Return(folly::unit)); + EXPECT_CALL(*upstream_, fetch(_, _)) + .WillOnce([this](Fetch fetch, std::shared_ptr 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(FetchOk{ + 0, + GroupOrder::OldestFirst, + false, + AbsoluteLocation{5, 20}, + }); + return folly::coro::makeTask( + 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>(); + 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 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(FetchOk{ + 0, + GroupOrder::NewestFirst, + false, + AbsoluteLocation{3, 5}, + }); + return folly::coro::makeTask( + 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>(); + ON_CALL(*laterConsumer, endOfFetch()).WillByDefault(Return(folly::unit)); + EXPECT_CALL(*upstream_, fetch(_, _)) + .WillOnce([this](Fetch fetch, std::shared_ptr 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(FetchOk{ + 0, + GroupOrder::OldestFirst, + false, + AbsoluteLocation{3, 1}, + }); + return folly::coro::makeTask( + 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>(); + 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 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(FetchOk{ + 0, + GroupOrder::NewestFirst, + false, + AbsoluteLocation{3, 3}, + }); + return folly::coro::makeTask( + 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>(); + 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) { diff --git a/moxygen/test/MoQLocationTest.cpp b/moxygen/test/MoQLocationTest.cpp index 16a40cce0..73a3ea48e 100644 --- a/moxygen/test/MoQLocationTest.cpp +++ b/moxygen/test/MoQLocationTest.cpp @@ -121,3 +121,45 @@ TEST(Location, Compare) { EXPECT_TRUE(loc2 < loc3); EXPECT_TRUE(loc3 > loc2); } + +TEST(Location, Next) { + // Normal increment + EXPECT_EQ((AbsoluteLocation{0, 0}.next()), (AbsoluteLocation{0, 1})); + EXPECT_EQ((AbsoluteLocation{5, 100}.next()), (AbsoluteLocation{5, 101})); + + // Object overflow -> next group + EXPECT_EQ( + (AbsoluteLocation{5, kEightByteLimit}.next()), (AbsoluteLocation{6, 0})); + + // Max location has no successor + EXPECT_FALSE(kLocationMax.next().has_value()); +} + +TEST(Location, Prev) { + // Normal decrement + EXPECT_EQ((AbsoluteLocation{0, 1}.prev()), (AbsoluteLocation{0, 0})); + EXPECT_EQ((AbsoluteLocation{5, 100}.prev()), (AbsoluteLocation{5, 99})); + + // Object underflow -> previous group + EXPECT_EQ( + (AbsoluteLocation{5, 0}.prev()), (AbsoluteLocation{4, kEightByteLimit})); + + // Min location has no predecessor + EXPECT_FALSE(kLocationMin.prev().has_value()); +} + +TEST(Location, NextInGroup) { + EXPECT_EQ((AbsoluteLocation{5, 10}.nextInGroup()), (AbsoluteLocation{5, 11})); + EXPECT_FALSE( + (AbsoluteLocation{5, kEightByteLimit}.nextInGroup().has_value())); +} + +TEST(Location, NextGroup) { + EXPECT_EQ((AbsoluteLocation{5, 10}.nextGroup()), (AbsoluteLocation{6, 0})); + EXPECT_FALSE((AbsoluteLocation{kEightByteLimit, 0}.nextGroup().has_value())); +} + +TEST(Location, PrevGroup) { + EXPECT_EQ((AbsoluteLocation{5, 10}.prevGroup()), (AbsoluteLocation{4, 0})); + EXPECT_FALSE((AbsoluteLocation{0, 10}.prevGroup().has_value())); +} diff --git a/moxygen/util/CMakeLists.txt b/moxygen/util/CMakeLists.txt index d582c0c14..3989de5e9 100644 --- a/moxygen/util/CMakeLists.txt +++ b/moxygen/util/CMakeLists.txt @@ -41,3 +41,12 @@ moxygen_add_library(moxygen_util_signal_handler Folly::folly_io_async_async_signal_handler Folly::folly_logging_logging ) + +moxygen_add_library(moxygen_util_location_interval_set + SRCS + LocationIntervalSet.cpp + EXPORTED_DEPS + moxygen_moq_framer +) + +add_subdirectory(test) diff --git a/moxygen/util/LocationIntervalSet.cpp b/moxygen/util/LocationIntervalSet.cpp new file mode 100644 index 000000000..9ab8b3d75 --- /dev/null +++ b/moxygen/util/LocationIntervalSet.cpp @@ -0,0 +1,88 @@ +/* + * Copyright (c) Meta Platforms, Inc. and affiliates. + * + * This source code is licensed under the MIT license found in the LICENSE + * file in the root directory of this source tree. + * + */ + +#include + +namespace moxygen { + +void LocationIntervalSet::insert(AbsoluteLocation start, AbsoluteLocation end) { + if (end < start) { + return; + } + + auto mergeStart = start; + auto mergeEnd = end; + + auto prevStart = start.prev(); + auto nextEnd = end.next(); + + auto it = intervals_.lower_bound(start); + + // Check the preceding interval — its end may overlap or be adjacent. + if (it != intervals_.begin()) { + auto prev = std::prev(it); + if (prevStart ? prev->second >= *prevStart : prev->second >= start) { + it = prev; + } + } + + while (it != intervals_.end()) { + if (nextEnd ? it->first > *nextEnd : it->first > end) { + break; + } + if (it->first < mergeStart) { + mergeStart = it->first; + } + if (it->second > mergeEnd) { + mergeEnd = it->second; + } + it = intervals_.erase(it); + } + + intervals_[mergeStart] = mergeEnd; +} + +LocationIntervalSet::MapType::const_iterator +LocationIntervalSet::findContaining(AbsoluteLocation loc) const { + if (intervals_.empty()) { + return intervals_.end(); + } + auto it = intervals_.upper_bound(loc); + if (it == intervals_.begin()) { + return intervals_.end(); + } + --it; + if (loc <= it->second) { + return it; + } + return intervals_.end(); +} + +bool LocationIntervalSet::contains(AbsoluteLocation loc) const { + return findContaining(loc) != intervals_.end(); +} + +std::optional LocationIntervalSet::findIntervalEnd( + AbsoluteLocation loc) const { + auto it = findContaining(loc); + if (it != intervals_.end()) { + return it->second; + } + return std::nullopt; +} + +std::optional> +LocationIntervalSet::findInterval(AbsoluteLocation loc) const { + auto it = findContaining(loc); + if (it != intervals_.end()) { + return std::make_pair(it->first, it->second); + } + return std::nullopt; +} + +} // namespace moxygen diff --git a/moxygen/util/LocationIntervalSet.h b/moxygen/util/LocationIntervalSet.h new file mode 100644 index 000000000..1840a2c2b --- /dev/null +++ b/moxygen/util/LocationIntervalSet.h @@ -0,0 +1,95 @@ +/* + * Copyright (c) Meta Platforms, Inc. and affiliates. + * + * This source code is licensed under the MIT license found in the LICENSE + * file in the root directory of this source tree. + * + */ + +#pragma once + +#include +#include +#include +#include + +namespace moxygen { + +/** + * LocationIntervalSet stores disjoint intervals of AbsoluteLocation. + * + * This is a portable alternative to quic::IntervalSet that works with + * AbsoluteLocation (two uint64_t values) without requiring uint128_t. + * + * Key operations: + * - insert(start, end): Add an interval, merging with overlapping/adjacent + * - contains(loc): Check if a location is within any interval + * - findIntervalEnd(loc): Find the end of the interval containing loc + * + * Query operations are O(log n) where n is the number of intervals. + * Insert is O(log n + k) where k is the number of merged intervals. + */ +class LocationIntervalSet { + public: + /** + * Insert an interval [start, end] (inclusive on both ends). + * Automatically merges with overlapping or adjacent intervals. + */ + void insert(AbsoluteLocation start, AbsoluteLocation end); + + /** + * Insert a single position. Equivalent to insert(loc, loc). + */ + void insert(AbsoluteLocation loc) { + insert(loc, loc); + } + + /** + * Check if a location is contained in any interval. + */ + bool contains(AbsoluteLocation loc) const; + + /** + * If loc is in an interval, return the end of that interval. + * Otherwise return nullopt. + */ + std::optional findIntervalEnd(AbsoluteLocation loc) const; + + /** + * If loc is in an interval, return both start and end of that interval. + * Otherwise return nullopt. + */ + std::optional> findInterval( + AbsoluteLocation loc) const; + + /** + * Check if the set is empty. + */ + bool empty() const { + return intervals_.empty(); + } + + /** + * Clear all intervals. + */ + void clear() { + intervals_.clear(); + } + + /** + * Get the number of intervals (not positions). + */ + size_t size() const { + return intervals_.size(); + } + + private: + using MapType = std::map; + MapType::const_iterator findContaining(AbsoluteLocation loc) const; + + // Map from interval start to interval end (inclusive). + // Invariant: intervals are disjoint and non-adjacent. + MapType intervals_; +}; + +} // namespace moxygen diff --git a/moxygen/util/test/CMakeLists.txt b/moxygen/util/test/CMakeLists.txt new file mode 100644 index 000000000..9450bd846 --- /dev/null +++ b/moxygen/util/test/CMakeLists.txt @@ -0,0 +1,18 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +if(NOT BUILD_TESTS) + return() +endif() + +moxygen_add_test(TARGET LocationIntervalSetTests + PREFIX "LocationIntervalSet" + SOURCES + LocationIntervalSetTest.cpp + DEPENDS + moxygen_util_location_interval_set + testmain +) diff --git a/moxygen/util/test/LocationIntervalSetTest.cpp b/moxygen/util/test/LocationIntervalSetTest.cpp new file mode 100644 index 000000000..138cf1c31 --- /dev/null +++ b/moxygen/util/test/LocationIntervalSetTest.cpp @@ -0,0 +1,273 @@ +/* + * Copyright (c) Meta Platforms, Inc. and affiliates. + * This source code is licensed under the MIT license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include +#include + +using namespace moxygen; + +class LocationIntervalSetTest : public ::testing::Test { + protected: + LocationIntervalSet set_; +}; + +TEST_F(LocationIntervalSetTest, EmptySet) { + EXPECT_TRUE(set_.empty()); + EXPECT_EQ(set_.size(), 0); + EXPECT_FALSE(set_.contains({0, 0})); + EXPECT_FALSE(set_.findIntervalEnd({0, 0}).has_value()); +} + +TEST_F(LocationIntervalSetTest, InsertSinglePosition) { + set_.insert({0, 7}); + + EXPECT_EQ(set_.size(), 1); + EXPECT_FALSE(set_.contains({0, 6})); + EXPECT_TRUE(set_.contains({0, 7})); + EXPECT_FALSE(set_.contains({0, 8})); + + // Adjacent single-position inserts merge. + set_.insert({0, 8}); + EXPECT_EQ(set_.size(), 1); + EXPECT_TRUE(set_.contains({0, 8})); + EXPECT_EQ(set_.findIntervalEnd({0, 7}), (AbsoluteLocation{0, 8})); +} + +TEST_F(LocationIntervalSetTest, SingleInterval) { + set_.insert({0, 5}, {0, 10}); + + EXPECT_FALSE(set_.empty()); + EXPECT_EQ(set_.size(), 1); + + // Before interval + EXPECT_FALSE(set_.contains({0, 4})); + // In interval + EXPECT_TRUE(set_.contains({0, 5})); + EXPECT_TRUE(set_.contains({0, 7})); + EXPECT_TRUE(set_.contains({0, 10})); + // After interval + EXPECT_FALSE(set_.contains({0, 11})); +} + +TEST_F(LocationIntervalSetTest, FindIntervalEnd) { + set_.insert({0, 5}, {0, 10}); + + EXPECT_FALSE(set_.findIntervalEnd({0, 4}).has_value()); + EXPECT_EQ(set_.findIntervalEnd({0, 5}), (AbsoluteLocation{0, 10})); + EXPECT_EQ(set_.findIntervalEnd({0, 7}), (AbsoluteLocation{0, 10})); + EXPECT_EQ(set_.findIntervalEnd({0, 10}), (AbsoluteLocation{0, 10})); + EXPECT_FALSE(set_.findIntervalEnd({0, 11}).has_value()); +} + +TEST_F(LocationIntervalSetTest, FindInterval) { + set_.insert({0, 5}, {0, 10}); + + // Before interval + EXPECT_FALSE(set_.findInterval({0, 4}).has_value()); + + // At start of interval + auto result = set_.findInterval({0, 5}); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->first, (AbsoluteLocation{0, 5})); + EXPECT_EQ(result->second, (AbsoluteLocation{0, 10})); + + // In middle of interval + result = set_.findInterval({0, 7}); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->first, (AbsoluteLocation{0, 5})); + EXPECT_EQ(result->second, (AbsoluteLocation{0, 10})); + + // At end of interval + result = set_.findInterval({0, 10}); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->first, (AbsoluteLocation{0, 5})); + EXPECT_EQ(result->second, (AbsoluteLocation{0, 10})); + + // After interval + EXPECT_FALSE(set_.findInterval({0, 11}).has_value()); +} + +TEST_F(LocationIntervalSetTest, FindIntervalMultiple) { + set_.insert({0, 0}, {0, 5}); + set_.insert({0, 10}, {0, 15}); + + // First interval + auto result = set_.findInterval({0, 3}); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->first, (AbsoluteLocation{0, 0})); + EXPECT_EQ(result->second, (AbsoluteLocation{0, 5})); + + // Gap between intervals + EXPECT_FALSE(set_.findInterval({0, 7}).has_value()); + + // Second interval + result = set_.findInterval({0, 12}); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->first, (AbsoluteLocation{0, 10})); + EXPECT_EQ(result->second, (AbsoluteLocation{0, 15})); +} + +TEST_F(LocationIntervalSetTest, DisjointIntervals) { + set_.insert({0, 0}, {0, 5}); + set_.insert({0, 10}, {0, 15}); + + EXPECT_EQ(set_.size(), 2); + + EXPECT_TRUE(set_.contains({0, 0})); + EXPECT_TRUE(set_.contains({0, 5})); + EXPECT_FALSE(set_.contains({0, 6})); + EXPECT_FALSE(set_.contains({0, 9})); + EXPECT_TRUE(set_.contains({0, 10})); + EXPECT_TRUE(set_.contains({0, 15})); + EXPECT_FALSE(set_.contains({0, 16})); +} + +TEST_F(LocationIntervalSetTest, MergeOverlapping) { + set_.insert({0, 0}, {0, 10}); + set_.insert({0, 5}, {0, 15}); + + EXPECT_EQ(set_.size(), 1); + EXPECT_TRUE(set_.contains({0, 0})); + EXPECT_TRUE(set_.contains({0, 15})); + EXPECT_EQ(set_.findIntervalEnd({0, 0}), (AbsoluteLocation{0, 15})); +} + +TEST_F(LocationIntervalSetTest, MergeAdjacent) { + set_.insert({0, 0}, {0, 5}); + set_.insert({0, 6}, {0, 10}); + + // Should merge because {0, 5} and {0, 6} are adjacent + EXPECT_EQ(set_.size(), 1); + EXPECT_EQ(set_.findIntervalEnd({0, 0}), (AbsoluteLocation{0, 10})); +} + +TEST_F(LocationIntervalSetTest, MergeMultiple) { + set_.insert({0, 0}, {0, 5}); + set_.insert({0, 10}, {0, 15}); + set_.insert({0, 20}, {0, 25}); + + EXPECT_EQ(set_.size(), 3); + + // Insert interval that spans all three + set_.insert({0, 3}, {0, 22}); + + EXPECT_EQ(set_.size(), 1); + EXPECT_EQ(set_.findIntervalEnd({0, 0}), (AbsoluteLocation{0, 25})); +} + +TEST_F(LocationIntervalSetTest, CrossGroupInterval) { + // Interval spanning multiple groups + set_.insert({0, 100}, {2, 50}); + + EXPECT_TRUE(set_.contains({0, 100})); + EXPECT_TRUE(set_.contains({0, 1000})); + EXPECT_TRUE(set_.contains({1, 0})); + EXPECT_TRUE(set_.contains({1, 500})); + EXPECT_TRUE(set_.contains({2, 0})); + EXPECT_TRUE(set_.contains({2, 50})); + EXPECT_FALSE(set_.contains({2, 51})); + EXPECT_FALSE(set_.contains({0, 99})); +} + +TEST_F(LocationIntervalSetTest, LargeGap) { + // Test O(1) insertion of large gap (like 2^60 groups) + constexpr uint64_t kLargeGroup = 1ULL << 60; + + set_.insert({1, 0}, {kLargeGroup - 1, kEightByteLimit}); + + EXPECT_EQ(set_.size(), 1); + EXPECT_FALSE(set_.contains({0, 0})); + EXPECT_TRUE(set_.contains({1, 0})); + EXPECT_TRUE(set_.contains({kLargeGroup / 2, 0})); + EXPECT_TRUE(set_.contains({kLargeGroup - 1, 0})); + EXPECT_FALSE(set_.contains({kLargeGroup, 0})); +} + +TEST_F(LocationIntervalSetTest, Clear) { + set_.insert({0, 0}, {0, 10}); + EXPECT_FALSE(set_.empty()); + + set_.clear(); + EXPECT_TRUE(set_.empty()); + EXPECT_FALSE(set_.contains({0, 5})); +} + +TEST_F(LocationIntervalSetTest, InsertInvalidInterval) { + // end < start should be ignored + set_.insert({0, 10}, {0, 5}); + EXPECT_TRUE(set_.empty()); +} + +TEST_F(LocationIntervalSetTest, AdjacentAcrossGroups) { + // {0, kEightByteLimit} is adjacent to {1, 0} + set_.insert({0, 0}, {0, kEightByteLimit}); + set_.insert({1, 0}, {1, 10}); + + // Should merge + EXPECT_EQ(set_.size(), 1); + EXPECT_EQ(set_.findIntervalEnd({0, 0}), (AbsoluteLocation{1, 10})); +} + +TEST_F(LocationIntervalSetTest, InsertSameIntervalTwice) { + set_.insert({0, 5}, {0, 10}); + set_.insert({0, 5}, {0, 10}); + + EXPECT_EQ(set_.size(), 1); +} + +TEST_F(LocationIntervalSetTest, InsertSubInterval) { + set_.insert({0, 0}, {0, 20}); + set_.insert({0, 5}, {0, 10}); // Subset + + EXPECT_EQ(set_.size(), 1); + EXPECT_EQ(set_.findIntervalEnd({0, 0}), (AbsoluteLocation{0, 20})); +} + +TEST_F(LocationIntervalSetTest, InsertSuperInterval) { + set_.insert({0, 5}, {0, 10}); + set_.insert({0, 0}, {0, 20}); // Superset + + EXPECT_EQ(set_.size(), 1); + EXPECT_EQ(set_.findIntervalEnd({0, 0}), (AbsoluteLocation{0, 20})); +} + +// Tests for AbsoluteLocation helper methods + +TEST(AbsoluteLocationTest, PrevGroupEnd) { + // Normal case: group > 0 + auto result = AbsoluteLocation{5, 10}.prevGroupEnd(); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->group, 4); + EXPECT_EQ(result->object, kEightByteLimit); + + // Edge case: group == 0 (no previous group) + result = AbsoluteLocation{0, 10}.prevGroupEnd(); + EXPECT_FALSE(result.has_value()); + + // Object value doesn't affect result + result = AbsoluteLocation{3, 0}.prevGroupEnd(); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->group, 2); + EXPECT_EQ(result->object, kEightByteLimit); +} + +TEST(AbsoluteLocationTest, PrevInGroup) { + // Normal case: object > 0 + auto result = AbsoluteLocation{5, 10}.prevInGroup(); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->group, 5); + EXPECT_EQ(result->object, 9); + + // Edge case: object == 0 (no previous in group) + result = AbsoluteLocation{5, 0}.prevInGroup(); + EXPECT_FALSE(result.has_value()); + + // Group 0, object > 0 + result = AbsoluteLocation{0, 5}.prevInGroup(); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(result->group, 0); + EXPECT_EQ(result->object, 4); +}