kvserver,changefeeds,crosscluster: set per-consumer catchup scan limit

Currently, it is easily possible for a single slow rangefeed consumer
to acquire the entire catchup scan quota for a given store, preventing
any other consumers from advancing.

Here, we introduce the concept of a ConsumerID into the rangefeed
request. A ConsumerID is that it represents a logical rangefeed
consumer such as a changefeed or LDR stream. Such consumers may make
multiple MuxRangeFeed requests to a given node despite sharing the
same downstream consumer.

When per-consumer catchup scan limiting is enabled, no single consumer
is allowed to consumer more than 75% of a given store's capacity. If
no ConsumerID is specified, a random consumer ID is assigned to all
rangefeeds originating from a given MuxRangeFeed call.

In the long run, we need a more sophisticated approach to solve
this. This change is aimed to be a small improvement that solves the
most egregious case: a single slow consumer consuming the entire
quota. The goal of this change is an easily backportable feature,
however, it comes at the cost of adding yet-another mechanism
to the existing systems attempting to limit catchup scans:

1. Client-side rate limiter,
2. Store-level CatchupIter semaphore,
3. Elastic CPU rate limiting in the main CatchUpScan loop, and
4. Any limiting imposed now or in the future by virtue of including and
   admission header in the request.

Informs #132438

Epic: none
Release note: None

Author: stevendanna

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -514,6 +514,7 @@ func (ca *changeAggregator) makeKVFeedCfg(
 		Knobs:               ca.knobs.FeedKnobs,
 		ScopedTimers:        ca.sliMetrics.Timers,
 		MonitoringCfg:       monitoringCfg,
+		ConsumerID:          int64(ca.spec.JobID),
 	}, nil
 }
 

pkg/ccl/changefeedccl/kvfeed/kv_feed.go

@@ -93,6 +93,8 @@ type Config struct {
 	Knobs TestingKnobs
 
 	ScopedTimers *timers.ScopedTimers
+
+	ConsumerID int64
 }
 
 // Run will run the kvfeed. The feed runs synchronously and returns an
@@ -123,6 +125,7 @@ func Run(ctx context.Context, cfg Config) error {
 		cfg.Writer, cfg.Spans, cfg.CheckpointSpans, cfg.CheckpointTimestamp,
 		cfg.SchemaChangeEvents, cfg.SchemaChangePolicy,
 		cfg.NeedsInitialScan, cfg.WithDiff, cfg.WithFiltering,
+		cfg.ConsumerID,
 		cfg.InitialHighWater, cfg.EndTime,
 		cfg.Codec,
 		cfg.SchemaFeed,
@@ -248,6 +251,7 @@ type kvFeed struct {
 	withDiff            bool
 	withFiltering       bool
 	withInitialBackfill bool
+	consumerID          int64
 	initialHighWater    hlc.Timestamp
 	endTime             hlc.Timestamp
 	writer              kvevent.Writer
@@ -278,6 +282,7 @@ func newKVFeed(
 	schemaChangeEvents changefeedbase.SchemaChangeEventClass,
 	schemaChangePolicy changefeedbase.SchemaChangePolicy,
 	withInitialBackfill, withDiff, withFiltering bool,
+	consumerID int64,
 	initialHighWater hlc.Timestamp,
 	endTime hlc.Timestamp,
 	codec keys.SQLCodec,
@@ -297,6 +302,7 @@ func newKVFeed(
 		withInitialBackfill: withInitialBackfill,
 		withDiff:            withDiff,
 		withFiltering:       withFiltering,
+		consumerID:          consumerID,
 		initialHighWater:    initialHighWater,
 		endTime:             endTime,
 		schemaChangeEvents:  schemaChangeEvents,
@@ -585,6 +591,7 @@ func (f *kvFeed) runUntilTableEvent(ctx context.Context, resumeFrontier span.Fro
 		Frontier:      resumeFrontier.Frontier(),
 		WithDiff:      f.withDiff,
 		WithFiltering: f.withFiltering,
+		ConsumerID:    f.consumerID,
 		Knobs:         f.knobs,
 		Timers:        f.timers,
 		RangeObserver: f.rangeObserver,

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

@@ -144,6 +144,7 @@ func TestKVFeed(t *testing.T) {
 		f := newKVFeed(buf, tc.spans, tc.checkpoint, hlc.Timestamp{},
 			tc.schemaChangeEvents, tc.schemaChangePolicy,
 			tc.needsInitialScan, tc.withDiff, true, /* withFiltering */
+			0, /* consumerID */
 			tc.initialHighWater, tc.endTime,
 			codec,
 			tf, sf, rangefeedFactory(ref.run), bufferFactory,

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -31,6 +31,7 @@ type rangeFeedConfig struct {
 	Spans         []kvcoord.SpanTimePair
 	WithDiff      bool
 	WithFiltering bool
+	ConsumerID    int64
 	RangeObserver kvcoord.RangeObserver
 	Knobs         TestingKnobs
 	Timers        *timers.ScopedTimers

pkg/ccl/crosscluster/producer/event_stream.go

@@ -153,6 +153,7 @@ func (s *eventStream) Start(ctx context.Context, txn *kv.Txn) (retErr error) {
 		rangefeed.WithFrontierQuantized(quantize.Get(&s.execCfg.Settings.SV)),
 		rangefeed.WithOnValues(s.onValues),
 		rangefeed.WithDiff(s.spec.WithDiff),
+		rangefeed.WithConsumerID(int64(s.streamID)),
 		rangefeed.WithInvoker(func(fn func() error) error { return fn() }),
 		rangefeed.WithFiltering(s.spec.WithFiltering),
 	}

pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go

@@ -283,7 +283,7 @@ func (s *activeMuxRangeFeed) start(ctx context.Context, m *rangefeedMuxer) error
 
 		for !s.transport.IsExhausted() {
 			args := makeRangeFeedRequest(
-				s.Span, s.token.Desc().RangeID, m.cfg.overSystemTable, s.startAfter, m.cfg.withDiff, m.cfg.withFiltering, m.cfg.withMatchingOriginIDs)
+				s.Span, s.token.Desc().RangeID, m.cfg.overSystemTable, s.startAfter, m.cfg.withDiff, m.cfg.withFiltering, m.cfg.withMatchingOriginIDs, m.cfg.consumerID)
 			args.Replica = s.transport.NextReplica()
 			args.StreamID = streamID
 			s.ReplicaDescriptor = args.Replica

pkg/kv/kvclient/kvcoord/dist_sender_rangefeed.go

@@ -67,6 +67,7 @@ type rangeFeedConfig struct {
 	withMetadata          bool
 	withMatchingOriginIDs []uint32
 	rangeObserver         RangeObserver
+	consumerID            int64
 
 	knobs struct {
 		// onRangefeedEvent invoked on each rangefeed event.
@@ -138,6 +139,12 @@ func WithMetadata() RangeFeedOption {
 	})
 }
 
+func WithConsumerID(cid int64) RangeFeedOption {
+	return optionFunc(func(c *rangeFeedConfig) {
+		c.consumerID = cid
+	})
+}
+
 // SpanTimePair is a pair of span along with its starting time. The starting
 // time is exclusive, i.e. the first possible emitted event (including catchup
 // scans) will be at startAfter.Next().
@@ -620,6 +627,7 @@ func makeRangeFeedRequest(
 	withDiff bool,
 	withFiltering bool,
 	withMatchingOriginIDs []uint32,
+	consumerID int64,
 ) kvpb.RangeFeedRequest {
 	admissionPri := admissionpb.BulkNormalPri
 	if isSystemRange {
@@ -631,6 +639,7 @@ func makeRangeFeedRequest(
 			Timestamp: startAfter,
 			RangeID:   rangeID,
 		},
+		ConsumerID:            consumerID,
 		WithDiff:              withDiff,
 		WithFiltering:         withFiltering,
 		WithMatchingOriginIDs: withMatchingOriginIDs,

pkg/kv/kvclient/rangefeed/config.go

@@ -39,6 +39,7 @@ type config struct {
 	withDiff              bool
 	withFiltering         bool
 	withMatchingOriginIDs []uint32
+	consumerID            int64
 	onUnrecoverableError  OnUnrecoverableError
 	onCheckpoint          OnCheckpoint
 	frontierQuantize      time.Duration
@@ -159,6 +160,12 @@ func WithOriginIDsMatching(originIDs ...uint32) Option {
 	})
 }
 
+func WithConsumerID(cid int64) Option {
+	return optionFunc(func(c *config) {
+		c.consumerID = cid
+	})
+}
+
 // WithInvoker makes an option to invoke the rangefeed tasks such as running the
 // the client and processing events emitted by the client with a caller-supplied
 // function, which can make it easier to introspect into work done by a given

pkg/kv/kvclient/rangefeed/rangefeed.go

@@ -349,6 +349,7 @@ func (f *RangeFeed) run(ctx context.Context, frontier span.Frontier, resumeWithF
 	if f.onMetadata != nil {
 		rangefeedOpts = append(rangefeedOpts, kvcoord.WithMetadata())
 	}
+	rangefeedOpts = append(rangefeedOpts, kvcoord.WithConsumerID(f.consumerID))
 
 	for i := 0; r.Next(); i++ {
 		ts := frontier.Frontier()

pkg/kv/kvclient/rangefeed/rangefeed_external_test.go

@@ -1451,7 +1451,7 @@ func TestRangeFeedIntentResolutionRace(t *testing.T) {
 	}
 	eventC := make(chan *kvpb.RangeFeedEvent)
 	sink := newChannelSink(ctx, eventC)
-	require.NoError(t, s3.RangeFeed(sink.ctx, &req, sink)) // check if we've errored yet
+	require.NoError(t, s3.RangeFeed(sink.ctx, &req, sink, nil)) // check if we've errored yet
 	require.NoError(t, sink.Error())
 	t.Logf("started rangefeed on %s", repl3)
 
@@ -1876,3 +1876,99 @@ func TestRangeFeedMetadataAutoSplit(t *testing.T) {
 		}
 	})
 }
+
+// TestRangefeedCatchupStarvation tests that a single MuxRangefeed
+// call cannot starve other users. Note that starvation is still
+// possible if there are more than 2 consumers of a given range.
+func TestRangefeedCatchupStarvation(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testutils.RunValues(t, "feed_type", feedTypes, func(t *testing.T, rt rangefeedTestType) {
+		ctx := context.Background()
+		settings := cluster.MakeTestingClusterSettings()
+		kvserver.RangefeedUseBufferedSender.Override(ctx, &settings.SV, rt.useBufferedSender)
+		kvserver.RangefeedEnabled.Override(ctx, &settings.SV, true)
+		// Lower the limit to make it more likely to get starved.
+		kvserver.ConcurrentRangefeedItersLimit.Override(ctx, &settings.SV, 8)
+		kvserver.PerConsumerCatchupLimit.Override(ctx, &settings.SV, 6)
+		srv, _, db := serverutils.StartServer(t, base.TestServerArgs{
+			Settings: settings,
+		})
+		defer srv.Stopper().Stop(ctx)
+		s := srv.ApplicationLayer()
+		ts := s.Clock().Now()
+		scratchKey := append(s.Codec().TenantPrefix(), keys.ScratchRangeMin...)
+		scratchKey = scratchKey[:len(scratchKey):len(scratchKey)]
+		mkKey := func(k string) roachpb.Key {
+			return encoding.EncodeStringAscending(scratchKey, k)
+		}
+		ranges := 32
+		keysPerRange := 128
+		totalKeys := ranges * keysPerRange
+		for i := range ranges {
+			for j := range keysPerRange {
+				k := mkKey(fmt.Sprintf("%d-%d", i, j))
+				require.NoError(t, db.Put(ctx, k, 1))
+			}
+			_, _, err := srv.SplitRange(mkKey(fmt.Sprintf("%d", i)))
+			require.NoError(t, err)
+		}
+
+		span := roachpb.Span{Key: scratchKey, EndKey: scratchKey.PrefixEnd()}
+		f, err := rangefeed.NewFactory(s.AppStopper(), db, s.ClusterSettings(), nil)
+		require.NoError(t, err)
+
+		blocked := make(chan struct{})
+		r1, err := f.RangeFeed(ctx, "consumer-1-rf-1", []roachpb.Span{span}, ts,
+			func(ctx context.Context, value *kvpb.RangeFeedValue) {
+				blocked <- struct{}{}
+				<-ctx.Done()
+			},
+			rangefeed.WithConsumerID(1),
+		)
+		require.NoError(t, err)
+		defer r1.Close()
+		<-blocked
+
+		// Multiple rangefeeds from the same ConsumeID should
+		// be treated as the same consumer and thus they
+		// shouldn't be able to overwhelm the overall store
+		// quota.
+		for i := range 8 {
+			r1, err := f.RangeFeed(ctx, fmt.Sprintf("consumer-1-rf-%d", i+2), []roachpb.Span{span}, ts,
+				func(ctx context.Context, value *kvpb.RangeFeedValue) { <-ctx.Done() },
+				rangefeed.WithConsumerID(1),
+			)
+			require.NoError(t, err)
+			defer r1.Close()
+		}
+
+		// Despite 9 rangefeeds above each needing 32 catchup
+		// scans, the following rangefeed should always make
+		// progress because it has a different consumer ID.
+		r2ConsumedRow := make(chan roachpb.Key)
+		r2, err := f.RangeFeed(ctx, "rf2", []roachpb.Span{span}, ts,
+			func(ctx context.Context, value *kvpb.RangeFeedValue) {
+				r2ConsumedRow <- value.Key
+			},
+			rangefeed.WithConsumerID(2),
+		)
+		require.NoError(t, err)
+		defer r2.Close()
+
+		// Wait until we see every key we've writen on rf2.
+		seen := make(map[string]struct{}, 0)
+		for {
+			select {
+			case r := <-r2ConsumedRow:
+				seen[r.String()] = struct{}{}
+				if len(seen) >= totalKeys {
+					return
+				}
+			case <-time.After(testutils.DefaultSucceedsSoonDuration):
+				t.Fatal("test timed out")
+			}
+		}
+	})
+}