testserver: reduce wait for nodelocal capability

This adds the ability to restart the rangefeedcache used by the tenant
capability watcher and then issues such a restart before waiting for
the nodelocal capability.

The initial scan the restart forces typically takes less time than
waiting out the closed timestamp.

Epic: CRDB-18499

Release note: None

Author: stevendanna

pkg/kv/kvclient/rangefeed/rangefeedcache/watcher.go

@@ -77,6 +77,9 @@ type Watcher struct {
 
 	lastFrontierTS hlc.Timestamp // used to assert monotonicity across rangefeed attempts
 
+	// Used to force a restart during testing.
+	restartErrCh chan error
+
 	knobs TestingKnobs
 }
 
@@ -177,6 +180,7 @@ func NewWatcher(
 		withPrevValue:    withPrevValue,
 		translateEvent:   translateEvent,
 		onUpdate:         onUpdate,
+		restartErrCh:     make(chan error),
 	}
 	if knobs != nil {
 		w.knobs = *knobs
@@ -353,12 +357,24 @@ func (s *Watcher) Run(ctx context.Context) error {
 
 		case err := <-errCh:
 			return err
+		case err := <-s.restartErrCh:
+			return err
 		case err := <-s.knobs.ErrorInjectionCh:
 			return err
 		}
 	}
 }
 
+var restartErr = errors.New("testing restart requested")
+
+// TestingRestart injects an error into the rangefeed cache, forcing
+// it to restart. This is separate from the testing knob so that we
+// can force a restart from test infrastructure without overriding the
+// user-provided testing knobs.
+func (s *Watcher) TestingRestart() {
+	s.restartErrCh <- restartErr
+}
+
 func (s *Watcher) handleUpdate(
 	ctx context.Context, buffer *rangefeedbuffer.Buffer, ts hlc.Timestamp, updateType UpdateType,
 ) {

pkg/multitenant/tenantcapabilities/tenantcapabilitieswatcher/watcher.go

@@ -54,6 +54,10 @@ type Watcher struct {
 	startCh  chan struct{}
 	startErr error
 
+	// rfc provides access to the underlying
+	// rangefeedcache.Watcher for testing.
+	rfc *rangefeedcache.Watcher
+
 	// initialScan is used to synchronize the Start() method with the
 	// reception of the initial batch of values from the rangefeed
 	// (which happens asynchronously).
@@ -212,6 +216,7 @@ func (w *Watcher) startRangeFeed(ctx context.Context) error {
 	if err := rangefeedcache.Start(ctx, w.stopper, rfc, w.onError); err != nil {
 		return err
 	}
+	w.rfc = rfc
 
 	// Wait for the initial scan before returning.
 	select {
@@ -346,6 +351,12 @@ func (w *Watcher) handleIncrementalUpdate(
 	}
 }
 
+func (w *Watcher) TestingRestart() {
+	if w.rfc != nil {
+		w.rfc.TestingRestart()
+	}
+}
+
 // TestingFlushCapabilitiesState flushes the underlying global tenant capability
 // state for testing purposes. The returned entries are sorted by tenant ID.
 func (w *Watcher) TestingFlushCapabilitiesState() (entries []tenantcapabilities.Entry) {

pkg/server/testserver.go

@@ -1201,18 +1201,13 @@ func (ts *TestServer) StartTenant(
 	baseCfg.DisableTLSForHTTP = params.DisableTLSForHTTP
 	baseCfg.EnableDemoLoginEndpoint = params.EnableDemoLoginEndpoint
 
-	// Waiting for capabilities can take 3+ seconds since the
-	// rangefeedcache needs to wait for the closed timestamp
-	// before flushing updates. To avoid paying this cost in all
-	// cases, we only set the nodelocal storage capability if the
-	// caller has configured an ExternalIODir since nodelocal
-	// storage only works with that configured.
+	// Waiting for capabilities can time To avoid paying this cost in all
+	// cases, we only set the nodelocal storage capability if the caller has
+	// configured an ExternalIODir since nodelocal storage only works with
+	// that configured.
 	//
-	// TODO(ssd): We should do more here. We could have the caller
-	// pass in explicitly that they want these capabilities. Or,
-	// we could modify the system in some way to avoid waiting on
-	// capabilities for so long. Also, note that this doesn't
-	// apply to StartSharedProcessTenant.
+	// TODO(ssd): We do not set this capability in
+	// StartSharedProcessTenant.
 	shouldGrantNodelocalCap := ts.params.ExternalIODir != ""
 	canGrantNodelocalCap := ts.ClusterSettings().Version.IsActive(ctx, clusterversion.V23_1TenantCapabilities)
 	if canGrantNodelocalCap && shouldGrantNodelocalCap {
@@ -1225,6 +1220,11 @@ func (ts *TestServer) StartTenant(
 				return nil, err
 			}
 		} else {
+			// Restart the capabilities watcher. Restarting the
+			// watcher forces a new initial scan which is faster
+			// than waiting out the closed timestamp interval
+			// required to see new updates.
+			ts.tenantCapabilitiesWatcher.TestingRestart()
 			if err := testutils.SucceedsSoonError(func() error {
 				capabilities, found := ts.TenantCapabilitiesReader().GetCapabilities(params.TenantID)
 				if !found {