Merge #154717

154717: kvserver: return error from IntentScannerConstructor r=stevendanna a=stevendanna

I tried to do git archeology to determine why we didn't return an error here. I think it is generally a result of the previous processor architecture which would have been running this constructor in a different async task.

I'll note that if we _are_ currently seeing an error on this path, then it would potentially lead to a panic, so it seems unlikely we are seeing errors here often.

If we do see an error here, it is almost surely much better to abandon this register attempt and not attach the process to the replica.

My ulterior motive here is that we very much would like to be able to ensure that all messages sent to the rangefeed client flow through the same path (registration -> stream -> sender) as this helps reason about the correctness of changes.

Epic: none
Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/rangefeed/processor.go

@@ -135,18 +135,20 @@ type Processor interface {
 	// It is ok to start registering streams before background initialization
 	// completes.
 	//
-	// The provided iterator is used to initialize the rangefeed's resolved
-	// timestamp. It must obey the contract of an iterator used for an
-	// initResolvedTSScan. The Processor promises to clean up the iterator by
-	// calling its Close method when it is finished.
+	// The provided IntentScannerConstructor is used to construct a lock table
+	// iterator which will be used to initialize the rangefeed's resolved
+	// timestamp. It must be called under the same raftMu lock as first call to
+	// Register to ensure that there are no missing events.
 	//
-	// Note that newRtsIter must be called under the same lock as first
-	// registration to ensure that all there would be no missing events.
-	// This is currently achieved by Register function synchronizing with
-	// the work loop before the lock is released.
+	// If IntentScannerConstructor returns a non-nil error, the processor will be
+	// stopped. Otherwise, the intent scanner is successfully initialized. It must
+	// obey the contract of an iterator used for an initResolvedTSScan. The
+	// processor promises to clean up the iterator by calling its Close method
+	// when it's finished.
 	//
-	// If the iterator is nil then no initialization scan will be performed and
-	// the resolved timestamp will immediately be considered initialized.
+	// If the IntentScannerConstructor is nil then no initialization scan will be
+	// performed and the resolved timestamp will immediately be considered
+	// initialized. This should only be the case in tests.
 	Start(stopper *stop.Stopper, newRtsIter IntentScannerConstructor) error
 	// Stop processor and close all registrations.
 	//
@@ -306,4 +308,4 @@ type logicalOpMetadata struct {
 // IntentScannerConstructor is used to construct an IntentScanner. It
 // should be called from underneath a stopper task to ensure that the
 // engine has not been closed.
-type IntentScannerConstructor func() IntentScanner
+type IntentScannerConstructor func() (IntentScanner, error)

pkg/kv/kvserver/rangefeed/processor_helpers_test.go

@@ -311,8 +311,8 @@ func withMetrics(m *Metrics) option {
 func withRtsScanner(scanner IntentScanner) option {
 	return func(config *testConfig) {
 		if scanner != nil {
-			config.isc = func() IntentScanner {
-				return scanner
+			config.isc = func() (IntentScanner, error) {
+				return scanner, nil
 			}
 		}
 	}

pkg/kv/kvserver/rangefeed/processor_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
+	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
@@ -1582,3 +1583,52 @@ func TestProcessorContextCancellation(t *testing.T) {
 		}
 	})
 }
+
+// TestIntentScannerOnError tests that when a processor is given with an intent
+// scanner constructor that fails to create a scanner, the processor will fail
+// to start gracefully.
+func TestIntentScannerOnError(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	ctx := context.Background()
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+
+	cfg := testConfig{
+		Config: Config{
+			RangeID:  2,
+			Stopper:  stopper,
+			Span:     roachpb.RSpan{Key: roachpb.RKey("a"), EndKey: roachpb.RKey("z")},
+			Metrics:  NewMetrics(),
+			Priority: true,
+		},
+	}
+	sch := NewScheduler(SchedulerConfig{
+		Workers:         1,
+		PriorityWorkers: 1,
+		Metrics:         NewSchedulerMetrics(time.Second),
+	})
+	require.NoError(t, sch.Start(ctx, stopper))
+	cfg.Scheduler = sch
+
+	s := NewProcessor(cfg.Config)
+	erroringScanConstructor := func() (IntentScanner, error) {
+		return nil, errors.New("scanner error")
+	}
+	err := s.Start(stopper, erroringScanConstructor)
+	require.ErrorContains(t, err, "scanner error")
+
+	// The processor should be stopped eventually.
+	p := (s).(*ScheduledProcessor)
+	testutils.SucceedsSoon(t, func() error {
+		select {
+		case <-p.stoppedC:
+			_, ok := sch.shards[shardIndex(p.ID(), len(sch.shards), p.Priority)].procs[p.ID()]
+			require.False(t, ok)
+			require.False(t, sch.priorityIDs.Contains(p.ID()))
+			return nil
+		default:
+			return errors.New("processor not stopped")
+		}
+	})
+}

pkg/kv/kvserver/rangefeed/scheduled_processor.go

@@ -128,17 +128,24 @@ func (p *ScheduledProcessor) Start(
 	// Launch an async task to scan over the resolved timestamp iterator and
 	// initialize the unresolvedIntentQueue.
 	if rtsIterFunc != nil {
-		rtsIter := rtsIterFunc()
+		rtsIter, err := rtsIterFunc()
+		if err != nil {
+			// No need to close rtsIter if error is non-nil.
+			p.scheduler.StopProcessor()
+			return err
+		}
+
 		initScan := newInitResolvedTSScan(p.Span, p, rtsIter)
 		// TODO(oleg): we need to cap number of tasks that we can fire up across
 		// all feeds as they could potentially generate O(n) tasks during start.
-		err := stopper.RunAsyncTask(p.taskCtx, "rangefeed: init resolved ts", initScan.Run)
+		err = stopper.RunAsyncTask(p.taskCtx, "rangefeed: init resolved ts", initScan.Run)
 		if err != nil {
 			initScan.Cancel()
 			p.scheduler.StopProcessor()
 			return err
 		}
 	} else {
+		// This case should only be reached in tests.
 		p.initResolvedTS(p.taskCtx, nil)
 	}
 

pkg/kv/kvserver/replica_rangefeed.go

@@ -514,19 +514,13 @@ func (r *Replica) registerWithRangefeedRaftMuLocked(
 	p = rangefeed.NewProcessor(cfg)
 
 	// Start it with an iterator to initialize the resolved timestamp.
-	rtsIter := func() rangefeed.IntentScanner {
+	rtsIter := func() (rangefeed.IntentScanner, error) {
 		// Assert that we still hold the raftMu when this is called to ensure
 		// that the rtsIter reads from the current snapshot. The replica
 		// synchronizes with the rangefeed Processor calling this function by
 		// waiting for the Register call below to return.
 		r.raftMu.AssertHeld()
-
-		scanner, err := rangefeed.NewSeparatedIntentScanner(streamCtx, r.store.TODOEngine(), desc.RSpan())
-		if err != nil {
-			stream.SendError(kvpb.NewError(err))
-			return nil
-		}
-		return scanner
+		return rangefeed.NewSeparatedIntentScanner(streamCtx, r.store.TODOEngine(), desc.RSpan())
 	}
 
 	// NB: This only errors if the stopper is stopping, and we have to return here

pkg/server/node.go

@@ -2308,6 +2308,8 @@ func (n *Node) muxRangeFeed(muxStream kvpb.RPCInternal_MuxRangeFeedStream) error
 			// stream manager. If rangefeed disconnects with an error after being
 			// successfully registered, it calls streamSink.SendError.
 			if disconnector, err := n.stores.RangeFeed(streamCtx, req, streamSink, limiter); err != nil {
+				// The rangefeed was not registered, so it should be safe to send this
+				// error directly to the stream rather than via the registration.
 				streamSink.SendError(kvpb.NewError(err))
 			} else {
 				sm.AddStream(req.StreamID, disconnector)