kvserver/rangefeed: remove future package

Previously, to reduce O(ranges) goroutines and avoid blocking on rangefeed
completion, we introduced the future package to manage rangefeed completion by
invoking a future callback when an error occurred. The future package makes it
difficult to justify which goroutine the callback runs on and which locks are
being held. This patch introduces a new approach, replacing the future usage.
Each registration now uses the embedded StreamMuxer.DisconnectStreamWithError
method to signal rangefeed completion. StreamMuxer will manage the shutdown
logic and additional stream cleanup.

Part of: #126561
Release note: none

pkg/kv/kvclient/rangefeed/BUILD.bazel

@@ -87,7 +87,6 @@ go_test(
         "//pkg/testutils/testcluster",
         "//pkg/util/ctxgroup",
         "//pkg/util/encoding",
-        "//pkg/util/future",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",

pkg/kv/kvclient/rangefeed/rangefeed_external_test.go

@@ -40,7 +40,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/storageutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
-	"github.com/cockroachdb/cockroach/pkg/util/future"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -1463,8 +1462,8 @@ func TestRangeFeedIntentResolutionRace(t *testing.T) {
 	}
 	eventC := make(chan *kvpb.RangeFeedEvent)
 	sink := newChannelSink(ctx, eventC)
-	fErr := future.MakeAwaitableFuture(s3.RangeFeed(&req, sink))
-	require.NoError(t, fErr.Get()) // check if we've errored yet
+	require.NoError(t, s3.RangeFeed(&req, sink)) // check if we've errored yet
+	require.NoError(t, sink.Error())
 	t.Logf("started rangefeed on %s", repl3)
 
 	// Spawn a rangefeed monitor, which posts checkpoint updates to checkpointC.
@@ -1621,17 +1620,18 @@ func TestRangeFeedIntentResolutionRace(t *testing.T) {
 	}
 
 	// The rangefeed should still be running.
-	require.NoError(t, fErr.Get())
+	require.NoError(t, sink.Error())
 }
 
 // channelSink is a rangefeed sink which posts events to a channel.
 type channelSink struct {
-	ctx context.Context
-	ch  chan<- *kvpb.RangeFeedEvent
+	ctx  context.Context
+	ch   chan<- *kvpb.RangeFeedEvent
+	done chan *kvpb.Error
 }
 
 func newChannelSink(ctx context.Context, ch chan<- *kvpb.RangeFeedEvent) *channelSink {
-	return &channelSink{ctx: ctx, ch: ch}
+	return &channelSink{ctx: ctx, ch: ch, done: make(chan *kvpb.Error, 1)}
 }
 
 func (c *channelSink) Context() context.Context {
@@ -1649,6 +1649,23 @@ func (c *channelSink) Send(e *kvpb.RangeFeedEvent) error {
 	}
 }
 
+// Error returns the error sent to the done channel if the sink has been
+// disconnected. It returns nil otherwise.
+func (c *channelSink) Error() error {
+	select {
+	case err := <-c.done:
+		return err.GoError()
+	default:
+		return nil
+	}
+}
+
+// Disconnect implements the Stream interface. It mocks the disconnect behavior
+// by sending the error to the done channel.
+func (c *channelSink) Disconnect(err *kvpb.Error) {
+	c.done <- err
+}
+
 // TestRangeFeedMetadataManualSplit tests that a spawned rangefeed emits a
 // metadata event which indicates if it spawned to due a manual split. The
 // test specifically conducts the following:

pkg/kv/kvserver/BUILD.bazel

@@ -201,7 +201,6 @@ go_library(
         "//pkg/util/encoding",
         "//pkg/util/envutil",
         "//pkg/util/errorutil",
-        "//pkg/util/future",
         "//pkg/util/growstack",
         "//pkg/util/grpcutil",
         "//pkg/util/grunning",
@@ -499,7 +498,6 @@ go_test(
         "//pkg/util/circuit",
         "//pkg/util/ctxgroup",
         "//pkg/util/encoding",
-        "//pkg/util/future",
         "//pkg/util/grunning",
         "//pkg/util/hlc",
         "//pkg/util/humanizeutil",

pkg/kv/kvserver/client_replica_circuit_breaker_test.go

@@ -35,7 +35,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
 	"github.com/cockroachdb/cockroach/pkg/util/circuit"
-	"github.com/cockroachdb/cockroach/pkg/util/future"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -451,6 +450,16 @@ type dummyStream struct {
 	name string
 	ctx  context.Context
 	recv chan *kvpb.RangeFeedEvent
+	done chan *kvpb.Error
+}
+
+func newDummyStream(ctx context.Context, name string) *dummyStream {
+	return &dummyStream{
+		ctx:  ctx,
+		name: name,
+		recv: make(chan *kvpb.RangeFeedEvent),
+		done: make(chan *kvpb.Error, 1),
+	}
 }
 
 func (s *dummyStream) Context() context.Context {
@@ -472,21 +481,34 @@ func (s *dummyStream) Send(ev *kvpb.RangeFeedEvent) error {
 	}
 }
 
+// Disconnect implements the Stream interface. It mocks the disconnect behavior
+// by sending the error to the done channel.
+func (s *dummyStream) Disconnect(err *kvpb.Error) {
+	s.done <- err
+}
+
 func waitReplicaRangeFeed(
-	ctx context.Context,
-	r *kvserver.Replica,
-	req *kvpb.RangeFeedRequest,
-	stream kvpb.RangeFeedEventSink,
+	ctx context.Context, r *kvserver.Replica, req *kvpb.RangeFeedRequest, stream *dummyStream,
 ) error {
-	rfErr, ctxErr := future.Wait(ctx, r.RangeFeed(req, stream, nil /* pacer */))
-	if ctxErr != nil {
-		return ctxErr
+	sendErrToStream := func(err *kvpb.Error) error {
+		var event kvpb.RangeFeedEvent
+		event.SetValue(&kvpb.RangeFeedError{
+			Error: *err,
+		})
+		return stream.Send(&event)
+	}
+
+	err := r.RangeFeed(req, stream, nil /* pacer */)
+	if err != nil {
+		return sendErrToStream(kvpb.NewError(err))
+	}
+
+	select {
+	case err := <-stream.done:
+		return sendErrToStream(err)
+	case <-ctx.Done():
+		return ctx.Err()
 	}
-	var event kvpb.RangeFeedEvent
-	event.SetValue(&kvpb.RangeFeedError{
-		Error: *kvpb.NewError(rfErr),
-	})
-	return stream.Send(&event)
 }
 
 // This test verifies that RangeFeed bypasses the circuit breaker. When the
@@ -508,7 +530,7 @@ func TestReplicaCircuitBreaker_RangeFeed(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	stream1 := &dummyStream{ctx: ctx, name: "rangefeed1", recv: make(chan *kvpb.RangeFeedEvent)}
+	stream1 := newDummyStream(ctx, "rangefeed1")
 	require.NoError(t, tc.Stopper().RunAsyncTask(ctx, "stream1", func(ctx context.Context) {
 		err := waitReplicaRangeFeed(ctx, tc.repls[0].Replica, args, stream1)
 		if ctx.Err() != nil {
@@ -562,7 +584,7 @@ func TestReplicaCircuitBreaker_RangeFeed(t *testing.T) {
 
 	// Start another stream during the "outage" to make sure it isn't rejected by
 	// the breaker.
-	stream2 := &dummyStream{ctx: ctx, name: "rangefeed2", recv: make(chan *kvpb.RangeFeedEvent)}
+	stream2 := newDummyStream(ctx, "rangefeed2")
 	require.NoError(t, tc.Stopper().RunAsyncTask(ctx, "stream2", func(ctx context.Context) {
 		err := waitReplicaRangeFeed(ctx, tc.repls[0].Replica, args, stream2)
 		if ctx.Err() != nil {

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -37,7 +37,6 @@ go_library(
         "//pkg/util/buildutil",
         "//pkg/util/container/heap",
         "//pkg/util/envutil",
-        "//pkg/util/future",
         "//pkg/util/hlc",
         "//pkg/util/interval",
         "//pkg/util/log",
@@ -89,7 +88,6 @@ go_test(
         "//pkg/testutils/skip",
         "//pkg/testutils/storageutils",
         "//pkg/util/encoding",
-        "//pkg/util/future",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",

pkg/kv/kvserver/rangefeed/bench_test.go

@@ -22,7 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/isolation"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
-	"github.com/cockroachdb/cockroach/pkg/util/future"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/stretchr/testify/require"
@@ -100,19 +100,17 @@ func runBenchmarkRangefeed(b *testing.B, opts benchmarkRangefeedOpts) {
 
 	// Add registrations.
 	streams := make([]*noopStream, opts.numRegistrations)
-	futures := make([]*future.ErrorFuture, opts.numRegistrations)
 	for i := 0; i < opts.numRegistrations; i++ {
 		// withDiff does not matter for these benchmarks, since the previous value
 		// is fetched and populated during Raft application.
 		const withDiff = false
 		// withFiltering does not matter for these benchmarks because doesn't fetch
 		// extra data.
 		const withFiltering = false
-		streams[i] = &noopStream{ctx: ctx}
-		futures[i] = &future.ErrorFuture{}
+		streams[i] = &noopStream{ctx: ctx, done: make(chan *kvpb.Error, 1)}
 		ok, _ := p.Register(span, hlc.MinTimestamp, nil,
 			withDiff, withFiltering, false, /* withOmitRemote */
-			streams[i], nil, futures[i])
+			streams[i], nil)
 		require.True(b, ok)
 	}
 
@@ -185,10 +183,9 @@ func runBenchmarkRangefeed(b *testing.B, opts benchmarkRangefeedOpts) {
 	b.StopTimer()
 	p.Stop()
 
-	for i, f := range futures {
-		regErr, err := future.Wait(ctx, f)
-		require.NoError(b, err)
-		require.NoError(b, regErr)
+	for i, s := range streams {
+		// p.Stop() sends a nil error to all streams to signal completion.
+		require.NoError(b, s.WaitForError(b))
 		require.Equal(b, b.N, streams[i].events-1) // ignore checkpoint after catchup
 	}
 }
@@ -197,6 +194,7 @@ func runBenchmarkRangefeed(b *testing.B, opts benchmarkRangefeedOpts) {
 type noopStream struct {
 	ctx    context.Context
 	events int
+	done   chan *kvpb.Error
 }
 
 func (s *noopStream) Context() context.Context {
@@ -211,3 +209,22 @@ func (s *noopStream) Send(*kvpb.RangeFeedEvent) error {
 // Note that Send itself is not thread-safe, but it is written to be used only
 // in a single threaded environment in this test, ensuring thread-safety.
 func (s *noopStream) SendIsThreadSafe() {}
+
+// Disconnect implements the Stream interface. It mocks the disconnect behavior
+// by sending the error to the done channel.
+func (s *noopStream) Disconnect(error *kvpb.Error) {
+	s.done <- error
+}
+
+// WaitForError waits for the rangefeed to complete and returns the error sent
+// to the done channel. It fails the test if rangefeed cannot complete within 30
+// seconds.
+func (s *noopStream) WaitForError(b *testing.B) error {
+	select {
+	case err := <-s.done:
+		return err.GoError()
+	case <-time.After(testutils.DefaultSucceedsSoonDuration):
+		b.Fatalf("time out waiting for rangefeed completion")
+		return nil
+	}
+}

pkg/kv/kvserver/rangefeed/processor.go

@@ -22,7 +22,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
-	"github.com/cockroachdb/cockroach/pkg/util/future"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
@@ -210,7 +209,6 @@ type Processor interface {
 		withOmitRemote bool,
 		stream Stream,
 		disconnectFn func(),
-		done *future.ErrorFuture,
 	) (bool, *Filter)
 	// DisconnectSpanWithErr disconnects all rangefeed registrations that overlap
 	// the given span with the given error.
@@ -595,7 +593,6 @@ func (p *LegacyProcessor) Register(
 	withOmitRemote bool,
 	stream Stream,
 	disconnectFn func(),
-	done *future.ErrorFuture,
 ) (bool, *Filter) {
 	// Synchronize the event channel so that this registration doesn't see any
 	// events that were consumed before this registration was called. Instead,
@@ -605,7 +602,7 @@ func (p *LegacyProcessor) Register(
 	blockWhenFull := p.Config.EventChanTimeout == 0 // for testing
 	r := newRegistration(
 		span.AsRawSpanWithNoLocals(), startTS, catchUpIter, withDiff, withFiltering, withOmitRemote,
-		p.Config.EventChanCap, blockWhenFull, p.Metrics, stream, disconnectFn, done,
+		p.Config.EventChanCap, blockWhenFull, p.Metrics, stream, disconnectFn,
 	)
 	select {
 	case p.regC <- r: