rangefeed: remove LegacyProcessor

The ScheduledProcessor has been in default use since 24.1,
substantially out performs the LegacyProcessor, and is the likely
place that new improvements will be implemented.

Here, we remove the LegacyProcessor, retiring the associated cluster
setting.

This is similar to cockroachdb#114410, but reconstructed from scratch.

Epic: none

Release note (ops change): The setting kv.rangefeed.scheduler.enabled
has been retired. The rangefeed scheduler is now unconditionally
enabled.

Co-authored-by: Erik Grinaker <[email protected]>

pkg/kv/kvserver/BUILD.bazel

@@ -395,7 +395,6 @@ go_test(
         "//pkg/kv",
         "//pkg/kv/kvclient",
         "//pkg/kv/kvclient/kvcoord",
-        "//pkg/kv/kvclient/rangefeed",
         "//pkg/kv/kvclient/rangefeed/rangefeedcache",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/abortspan",

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -19,7 +19,6 @@ go_library(
         "stream.go",
         "task.go",
         "test_helpers.go",
-        "testutil.go",
         "unbuffered_sender.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/rangefeed",

pkg/kv/kvserver/rangefeed/bench_test.go

@@ -24,7 +24,6 @@ import (
 )
 
 type benchmarkRangefeedOpts struct {
-	procType         procType
 	opType           opType
 	numRegistrations int
 	budget           int64
@@ -41,19 +40,16 @@ const (
 // BenchmarkRangefeed benchmarks the processor and registrations, by submitting
 // a set of events and waiting until they are all emitted.
 func BenchmarkRangefeed(b *testing.B) {
-	for _, procType := range testTypes {
-		for _, opType := range []opType{writeOpType, commitOpType, closedTSOpType} {
-			for _, numRegistrations := range []int{1, 10, 100} {
-				name := fmt.Sprintf("procType=%s/opType=%s/numRegs=%d", procType, opType, numRegistrations)
-				b.Run(name, func(b *testing.B) {
-					runBenchmarkRangefeed(b, benchmarkRangefeedOpts{
-						procType:         procType,
-						opType:           opType,
-						numRegistrations: numRegistrations,
-						budget:           math.MaxInt64,
-					})
+	for _, opType := range []opType{writeOpType, commitOpType, closedTSOpType} {
+		for _, numRegistrations := range []int{1, 10, 100} {
+			name := fmt.Sprintf("opType=%s/numRegs=%d", opType, numRegistrations)
+			b.Run(name, func(b *testing.B) {
+				runBenchmarkRangefeed(b, benchmarkRangefeedOpts{
+					opType:           opType,
+					numRegistrations: numRegistrations,
+					budget:           math.MaxInt64,
 				})
-			}
+			})
 		}
 	}
 }
@@ -90,7 +86,7 @@ func runBenchmarkRangefeed(b *testing.B, opts benchmarkRangefeedOpts) {
 	span := roachpb.RSpan{Key: roachpb.RKey("a"), EndKey: roachpb.RKey("z")}
 
 	p, h, stopper := newTestProcessor(b, withSpan(span), withBudget(budget), withChanCap(b.N),
-		withEventTimeout(time.Hour), withProcType(opts.procType))
+		withEventTimeout(time.Hour))
 	defer stopper.Stop(ctx)
 
 	// Add registrations.