Merge #110208 #110333

110208: roachtest: backup-restore/mixed-version use system db when querying job status r=rhu713 a=rhu713

Previously in `waitForJobSuccess()` in the backup-restore/mixed-version test, if the job was a full cluster restore that dropped `defaultdb`, the query to the jobs table would fail with `pq: database "defaultdb" is offline: restoring` as our connection was to `defaultdb`, which is dropped during full cluster restore. We fix this by adding an option to connect to a different database in roachtests and connecting to the system database in `waitForJobSuccess()` instead.

Fixes #110165

Release note: None

110333: rangefeed: omit txn push attempt when intent queue is empty r=erikgrinaker a=erikgrinaker

`hlc.Clock.Now()` mutex acquisition during rangefeed txn push attempts has been seen to be a significant source of contention when running with many rangefeeds.

This patch short-circuits the txn push attempt if we're not tracking any intents, which is the common case.

Epic: none
Release note: None

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

Author: 3 people

pkg/cmd/roachtest/cluster.go

@@ -2497,15 +2497,20 @@ func (c *clusterImpl) ConnE(
 		return nil, err
 	}
 
-	dataSourceName := urls[0]
+	u, err := url.Parse(urls[0])
+	if err != nil {
+		return nil, err
+	}
+
 	if connOptions.User != "" {
-		u, err := url.Parse(urls[0])
-		if err != nil {
-			return nil, err
-		}
 		u.User = url.User(connOptions.User)
-		dataSourceName = u.String()
 	}
+
+	if connOptions.DBName != "" {
+		u.Path = connOptions.DBName
+	}
+	dataSourceName := u.String()
+
 	if len(connOptions.Options) > 0 {
 		vals := make(url.Values)
 		for k, v := range connOptions.Options {

pkg/cmd/roachtest/option/connection_options.go

@@ -17,6 +17,7 @@ import (
 
 type ConnOption struct {
 	User       string
+	DBName     string
 	TenantName string
 	Options    map[string]string
 }
@@ -49,3 +50,9 @@ func ConnectTimeout(t time.Duration) func(*ConnOption) {
 	}
 	return ConnectionOption("connect_timeout", fmt.Sprintf("%d", sec))
 }
+
+func DBName(dbName string) func(*ConnOption) {
+	return func(option *ConnOption) {
+		option.DBName = dbName
+	}
+}

pkg/cmd/roachtest/tests/mixed_version_backup.go

@@ -45,6 +45,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/version"
+	"github.com/cockroachdb/errors"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -1361,19 +1362,30 @@ func (mvb *mixedVersionBackup) backupName(
 // error if the job doesn't succeed within the attempted retries.
 func (mvb *mixedVersionBackup) waitForJobSuccess(
 	ctx context.Context, l *logger.Logger, rng *rand.Rand, h *mixedversion.Helper, jobID int,
-) error {
+) (resErr error) {
 	var lastErr error
-	node, db := h.RandomDB(rng, mvb.roachNodes)
+	node := h.RandomNode(rng, mvb.roachNodes)
 	l.Printf("querying job status through node %d", node)
 
+	db, err := mvb.cluster.ConnE(ctx, l, node, option.DBName("system"))
+	if err != nil {
+		l.Printf("error connecting to node %d: %v", node, err)
+		return err
+	}
+	defer func() {
+		err := db.Close()
+		resErr = errors.CombineErrors(resErr, err)
+	}()
+
 	jobsQuery := "system.jobs WHERE id = $1"
 	if hasInternalSystemJobs(h) {
 		jobsQuery = fmt.Sprintf("(%s)", jobutils.InternalSystemJobsBaseQuery)
 	}
-	for r := retry.StartWithCtx(ctx, backupCompletionRetryOptions); r.Next(); {
+	r := retry.StartWithCtx(ctx, backupCompletionRetryOptions)
+	for r.Next() {
 		var status string
 		var payloadBytes []byte
-		err := db.QueryRow(
+		err := db.QueryRowContext(ctx,
 			fmt.Sprintf(`SELECT status, payload FROM %s`, jobsQuery), jobID,
 		).Scan(&status, &payloadBytes)
 		if err != nil {
@@ -1404,7 +1416,11 @@ func (mvb *mixedVersionBackup) waitForJobSuccess(
 		return nil
 	}
 
-	return fmt.Errorf("waiting for job to finish: %w", lastErr)
+	if r.CurrentAttempt() >= backupCompletionRetryOptions.MaxRetries {
+		return fmt.Errorf("exhausted all %d retries waiting for job %d to finish, last err: %w", backupCompletionRetryOptions.MaxRetries, jobID, lastErr)
+	}
+
+	return fmt.Errorf("error waiting for job to finish: %w", lastErr)
 }
 
 // computeTableContents will generate a list of `tableContents`

pkg/kv/kvserver/rangefeed/processor.go

@@ -462,9 +462,9 @@ func (p *LegacyProcessor) run(
 
 		// Check whether any unresolved intents need a push.
 		case <-txnPushTickerC:
-			// Don't perform transaction push attempts until the resolved
-			// timestamp has been initialized.
-			if !p.rts.IsInit() {
+			// Don't perform transaction push attempts until the resolved timestamp
+			// has been initialized, or if we're not tracking any intents.
+			if !p.rts.IsInit() || p.rts.intentQ.Len() == 0 {
 				continue
 			}
 

pkg/kv/kvserver/rangefeed/scheduled_processor.go

@@ -178,7 +178,9 @@ func (p *ScheduledProcessor) processEvents(ctx context.Context) {
 }
 
 func (p *ScheduledProcessor) processPushTxn(ctx context.Context) {
-	if !p.txnPushActive && p.rts.IsInit() {
+	// NB: Len() check avoids hlc.Clock.Now() mutex acquisition in the common
+	// case, which can be a significant source of contention.
+	if !p.txnPushActive && p.rts.IsInit() && p.rts.intentQ.Len() > 0 {
 		now := p.Clock.Now()
 		before := now.Add(-p.PushTxnsAge.Nanoseconds(), 0)
 		oldTxns := p.rts.intentQ.Before(before)