sql: stop retrying temp schema cleanup on poisoned transactions

Previously, the temporary schema cleanup job would retry up to five
times on any error, including those that irreversibly poisoned the
transaction. This caused unnecessary log noise (e.g., "txn already
encountered an error; cannot be used anymore").

This change improves the retry logic by detecting for
TxnAlreadyEncounteredErrorError and aborting immediately.

Fixes #158152
Epic: none
Release note (bug fix): Temporary schema cleanup no longer retries after poisoned transaction errors, reducing log noise.

Author: spilchen

pkg/sql/exec_util.go

@@ -2033,6 +2033,9 @@ type ExecutorTestingKnobs struct {
 	// OnTempObjectsCleanupDone will trigger when the temporary objects cleanup
 	// job is done.
 	OnTempObjectsCleanupDone func()
+	// TempObjectCleanupErrorInjection, if set, will be called during temp object
+	// cleanup and can return an error to inject into the cleanup process.
+	TempObjectCleanupErrorInjection func() error
 
 	// WithStatementTrace is called after the statement is executed in
 	// execStmtInOpenState.

pkg/sql/temporary_schema.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
@@ -447,30 +448,43 @@ func makeTemporaryObjectCleanerMetrics() *temporaryObjectCleanerMetrics {
 	}
 }
 
+// shouldStopTempObjectCleanupRetry returns true if the error indicates that we
+// should not retry the operation.
+func shouldStopTempObjectCleanupRetry(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	// Check if transaction is already poisoned. Once this happens, the
+	// transaction cannot be reused regardless of retry attempts.
+	return errors.HasType(err, (*kvpb.TxnAlreadyEncounteredErrorError)(nil))
+}
+
 // doTemporaryObjectCleanup performs the actual cleanup.
 func (c *TemporaryObjectCleaner) doTemporaryObjectCleanup(
 	ctx context.Context, closerCh <-chan struct{},
 ) error {
 	defer log.Dev.Infof(ctx, "completed temporary object cleanup job")
 	// Wrap the retry functionality with the default arguments.
 	retryFunc := func(ctx context.Context, do func() error) error {
-		return retry.WithMaxAttempts(
-			ctx,
-			retry.Options{
-				InitialBackoff: 1 * time.Second,
-				MaxBackoff:     1 * time.Minute,
-				Multiplier:     2,
-				Closer:         closerCh,
-			},
-			5, // maxAttempts
-			func() error {
-				err := do()
-				if err != nil {
-					log.Dev.Warningf(ctx, "error during schema cleanup, retrying: %v", err)
+		return retry.Options{
+			InitialBackoff: 1 * time.Second,
+			MaxBackoff:     1 * time.Minute,
+			Multiplier:     2,
+			MaxRetries:     4, // 5 total attempts (4 retries + 1 initial)
+			Closer:         closerCh,
+		}.DoWithRetryable(ctx, func(ctx context.Context) (bool, error) {
+			err := do()
+			if err != nil {
+				if shouldStopTempObjectCleanupRetry(err) {
+					log.Dev.Warningf(ctx, "error during schema cleanup, not retryable: %v", err)
+					return false, err // Don't retry
 				}
-				return err
-			},
-		)
+				log.Dev.Warningf(ctx, "error during schema cleanup, retrying: %v", err)
+				return true, err // Retry
+			}
+			return false, nil // Success
+		})
 	}
 
 	// For tenants, we will completely skip this logic since listing
@@ -507,6 +521,15 @@ func (c *TemporaryObjectCleaner) doTemporaryObjectCleanup(
 	log.Dev.Infof(ctx, "running temporary object cleanup background job")
 	var sessionIDs map[clusterunique.ID]struct{}
 	if err := c.db.DescsTxn(ctx, func(ctx context.Context, txn descs.Txn) error {
+		// Testing knob to inject errors during cleanup.
+		if c.testingKnobs.TempObjectCleanupErrorInjection != nil {
+			if err := retryFunc(ctx, func() (err error) {
+				return c.testingKnobs.TempObjectCleanupErrorInjection()
+			}); err != nil {
+				return err
+			}
+		}
+
 		sessionIDs = make(map[clusterunique.ID]struct{})
 		// Only see temporary schemas after some delay as safety
 		// mechanism.

pkg/sql/temporary_schema_test.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
@@ -22,8 +23,10 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
@@ -317,3 +320,84 @@ func constructNameToIDMapping(
 	}
 	return namesToID, tempSchemaNames
 }
+
+// TestTemporaryObjectCleanupRetriesWithPoisonedTransaction tests that the
+// cleanup process doesn't retry when encountering a poisoned transaction error.
+func TestTemporaryObjectCleanupRetriesWithPoisonedTransaction(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+
+	testCases := []struct {
+		name               string
+		errorInjectionFunc func(attemptCount *int) error
+		expectedAttempts   int
+		expectError        bool
+		errorSubstring     string
+	}{
+		{
+			name: "poisoned transaction stops retry",
+			errorInjectionFunc: func(attemptCount *int) error {
+				*attemptCount++
+				return &kvpb.TxnAlreadyEncounteredErrorError{}
+			},
+			expectedAttempts: 1,
+			expectError:      true,
+			errorSubstring:   "txn already encountered an error",
+		},
+		{
+			name: "regular error retries and succeeds",
+			errorInjectionFunc: func(attemptCount *int) error {
+				*attemptCount++
+				// Fail on first attempt, succeed on second
+				if *attemptCount == 1 {
+					return errors.New("some retryable error")
+				}
+				return nil
+			},
+			expectedAttempts: 2,
+			expectError:      false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var attemptCount int
+
+			testingKnobs := ExecutorTestingKnobs{
+				TempObjectCleanupErrorInjection: func() error {
+					return tc.errorInjectionFunc(&attemptCount)
+				},
+			}
+
+			s := serverutils.StartServerOnly(t, base.TestServerArgs{})
+			defer s.Stopper().Stop(ctx)
+
+			execCfg := s.ExecutorConfig().(ExecutorConfig)
+
+			// Create a temporary object cleaner with our testing knobs
+			cleaner := NewTemporaryObjectCleaner(
+				execCfg.Settings,
+				execCfg.InternalDB,
+				execCfg.Codec,
+				metric.NewRegistry(),
+				execCfg.SQLStatusServer,
+				func(context.Context, hlc.ClockTimestamp) (bool, error) { return true, nil },
+				testingKnobs,
+				nil,
+			)
+
+			// Run cleanup and verify behavior
+			err := cleaner.doTemporaryObjectCleanup(ctx, nil)
+			if tc.expectError {
+				require.Error(t, err)
+				require.ErrorContains(t, err, tc.errorSubstring)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, tc.expectedAttempts, attemptCount,
+				"expected %d attempts, got %d", tc.expectedAttempts, attemptCount)
+		})
+	}
+}