Merge cockroachdb#35466

35466: kv: stop DistSender from double-covering RangeFeeds during splits r=nvanbenschoten a=danhhz

If a RangeFeed was running over `[/a,/d)` and it split at `/b`, then
we'd get an error from the server and the span would be kicked out to
`divideAndSendRangeFeedToRanges`. The `RangeIterator` would usually hand
out the post-split `[/a,/b)` descriptor for `/a`, then advance to `/b`
and first try the pre-split `[/a,/d)` descriptor. Each would get a new
RangeFeed. The one over `[/a,/d)` would immediately come back with an
error from the server that it didn't fit in the bounds of a range, evict
the `RangeDescriptorCache` token, and get kicked out again to
`divideAndSendRangeFeedToRanges`. This time, because of the eviction,
the `RangeIterator` would get the post-split `[/b,/d)` descriptor. The
end result was one RangeFeed over `[/a,/b)` and two over `[/b,/d)`. A
second split at `/c` meant we could double it again and end up with 4
over `[/c,/d)`. RangeFeed always has the potential for sending
duplicates, so changefeeds have to be resilient to this, and this was
not a correctness issue, but it's obviously bad.

The fix is simple: use the same 'nextRS' trick that
`divideAndSendBatchToRanges` does to keep track of the uncovered part of
the input `rs` span to `divideAndSendRangeFeedToRanges` and use that to
trim the descriptors that come back.

Found when splits were added to TestChangefeedNemeses and manually
investiating why RangeFeed was returning duplicates. No test yet since
one is coming in cockroachdb#32721.

Release note: None

Co-authored-by: Daniel Harrison <[email protected]>

pkg/kv/dist_sender_rangefeed.go

@@ -86,13 +86,20 @@ func (ds *DistSender) divideAndSendRangeFeedToRanges(
 	rs roachpb.RSpan,
 	rangeCh chan<- singleRangeInfo,
 ) error {
+	// As RangeIterator iterates, it can return overlapping descriptors (and
+	// during splits, this happens frequently), but divideAndSendRangeFeedToRanges
+	// intends to split up the input into non-overlapping spans aligned to range
+	// boundaries. So, as we go, keep track of the remaining uncovered part of
+	// `rs` in `nextRS`.
+	nextRS := rs
 	ri := NewRangeIterator(ds)
-	for ri.Seek(ctx, rs.Key, Ascending); ri.Valid(); ri.Next(ctx) {
+	for ri.Seek(ctx, nextRS.Key, Ascending); ri.Valid(); ri.Next(ctx) {
 		desc := ri.Desc()
-		partialRS, err := rs.Intersect(desc)
+		partialRS, err := nextRS.Intersect(desc)
 		if err != nil {
 			return err
 		}
+		nextRS.Key = partialRS.EndKey
 		select {
 		case rangeCh <- singleRangeInfo{
 			desc:  desc,
@@ -102,7 +109,7 @@ func (ds *DistSender) divideAndSendRangeFeedToRanges(
 		case <-ctx.Done():
 			return ctx.Err()
 		}
-		if !ri.NeedAnother(rs) {
+		if !ri.NeedAnother(nextRS) {
 			break
 		}
 	}