settingswatcher: add version guard utility

Add the settingswatcher.VersionGuard utility. VersionGuard makes it
possible to check a cluster version in the context of a transaction.
Until a version gate is reached, VersionGuard will look up the version
set in the settings table. Once a version is reached, the guard can skip
the KV read because versions never go backwards.

The version guard will be used to migrate the sqlinstance, sql_liveness,
lease system tables to a regional by row compatible index. The in-memory
IsActive check is insufficient for the migration, because the setting
can change in between when the transaction starts and when the
transaction commits.

Part of cockroachdb#94843

pkg/server/settingswatcher/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "overrides.go",
         "row_decoder.go",
         "settings_watcher.go",
+        "version_guard.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/server/settingswatcher",
     visibility = ["//visibility:public"],
@@ -44,11 +45,13 @@ go_test(
         "main_test.go",
         "row_decoder_external_test.go",
         "settings_watcher_external_test.go",
+        "version_guard_test.go",
     ],
     args = ["-test.timeout=295s"],
     deps = [
         ":settingswatcher",
         "//pkg/base",
+        "//pkg/clusterversion",
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/kv/kvclient/rangefeed",

pkg/server/settingswatcher/version_guard.go

@@ -0,0 +1,54 @@
+package settingswatcher
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/kv"
+)
+
+// VersionGuard is a utility for checking the cluster version in a transaction.
+// VersionGuard is optimized to avoid the extra kv read overhead once the
+// cluster is finalized.
+//
+// Example Usage:
+// guard, err := watcher.MakeVersionGuard(ctx, txn, version.MaxVersionGateToCheck)
+//
+//	if err != nil {
+//	   return err // unable to read version
+//	}
+//
+//	if interlock.IsActive(version.SomeVersionLessThanMax) {
+//	  ...
+//	} else if interlock.IsActive(version.MaxVersionGateToCheck) {
+//
+//	  ...
+//	}
+type VersionGuard struct {
+	maxGateIsActive bool
+	txnVersion      clusterversion.ClusterVersion
+}
+
+// MakVersionGuard constructs a version guard for the transaction.
+func (s *SettingsWatcher) MakeVersionGuard(
+	ctx context.Context, txn *kv.Txn, maxGate clusterversion.Key,
+) (VersionGuard, error) {
+	if s.settings.Version.IsActive(ctx, maxGate) {
+		return VersionGuard{
+			maxGateIsActive: true,
+		}, nil
+	}
+	txnVersion, err := s.GetClusterVersionFromStorage(ctx, txn)
+	return VersionGuard{
+		txnVersion: txnVersion,
+	}, err
+}
+
+// IsActive returns true if the transaction should treat the version guard as
+// active.
+func (v *VersionGuard) IsActive(version clusterversion.Key) bool {
+	if v.maxGateIsActive {
+		return true
+	}
+	return v.txnVersion.IsActive(version)
+}

pkg/server/settingswatcher/version_guard_test.go

@@ -0,0 +1,117 @@
+package settingswatcher_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/server/settingswatcher"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/stretchr/testify/require"
+)
+
+func TestVersionGuard(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	ctx := context.Background()
+	s, sqlDB, kvDB := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(ctx)
+	tDB := sqlutils.MakeSQLRunner(sqlDB)
+
+	type testCase struct {
+		name            string
+		storageVersion  clusterversion.Key
+		settingsVersion clusterversion.Key
+		checkVersions   map[clusterversion.Key]bool
+	}
+
+	initialVersion := clusterversion.V22_2
+	maxVersion := clusterversion.V23_1
+	tests := []testCase{
+		{
+			name:            "unfinalized",
+			storageVersion:  initialVersion,
+			settingsVersion: initialVersion,
+			checkVersions: map[clusterversion.Key]bool{
+				initialVersion:            true,
+				clusterversion.V23_1Start: false,
+				maxVersion:                false,
+			},
+		},
+		{
+			name:            "mid-finalize",
+			storageVersion:  clusterversion.V23_1Start,
+			settingsVersion: initialVersion,
+			checkVersions: map[clusterversion.Key]bool{
+				initialVersion:            true,
+				clusterversion.V23_1Start: true,
+				maxVersion:                false,
+			},
+		},
+		{
+			name:            "finalized",
+			storageVersion:  maxVersion,
+			settingsVersion: maxVersion,
+			checkVersions: map[clusterversion.Key]bool{
+				initialVersion:            true,
+				clusterversion.V23_1Start: true,
+				maxVersion:                true,
+			},
+		},
+		{
+			// Once the version guard's max version is active, it no longer
+			// consults the stored value. This allows us to remove the overhead
+			// of the version guard after finalization completes. A storage
+			// version behind the settings version should not exist in
+			// production.
+			name:            "verify-optimization",
+			storageVersion:  initialVersion,
+			settingsVersion: maxVersion,
+			checkVersions: map[clusterversion.Key]bool{
+				initialVersion:            true,
+				maxVersion:                true,
+				clusterversion.V23_1Start: true,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			settings := cluster.MakeTestingClusterSettingsWithVersions(
+				clusterversion.ByKey(maxVersion),
+				clusterversion.ByKey(initialVersion),
+				false,
+			)
+			require.NoError(t, clusterversion.Initialize(ctx, clusterversion.ByKey(test.settingsVersion), &settings.SV))
+			settingVersion := clusterversion.ClusterVersion{Version: clusterversion.ByKey(test.settingsVersion)}
+			require.NoError(t, settings.Version.SetActiveVersion(ctx, settingVersion))
+
+			storageVersion := clusterversion.ClusterVersion{Version: clusterversion.ByKey(test.storageVersion)}
+			marshaledVersion, err := storageVersion.Marshal()
+
+			require.NoError(t, err)
+			tDB.Exec(t, `
+				UPDATE system.settings
+				SET value = $1
+				WHERE name = 'version'`, marshaledVersion)
+
+			watcher := settingswatcher.New(nil, s.Codec(), settings, nil, nil, nil)
+			require.NoError(t, kvDB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+				guard, err := watcher.MakeVersionGuard(ctx, txn, maxVersion)
+				if err != nil {
+					return err
+				}
+
+				for version, expect := range test.checkVersions {
+					require.Equal(t, expect, guard.IsActive(version), "expect guard.IsActive(%v) to be %t", version, expect)
+				}
+
+				return nil
+			}))
+		})
+	}
+}

pkg/server/testserver.go

@@ -811,6 +811,11 @@ func (t *TestTenant) Tracer() *tracing.Tracer {
 	return t.SQLServer.ambientCtx.Tracer
 }
 
+// SettingsWatcher is part of the TestTenantInterface.
+func (t *TestTenant) SettingsWatcher() interface {} {
+	return t.SQLServer.settingsWatcher
+}
+
 // WaitForTenantEndKeySplit is part of the TestTenantInterface.
 func (t *TestTenant) WaitForTenantEndKeySplit(ctx context.Context) error {
 	// Wait until the tenant end key split happens.
@@ -1192,6 +1197,11 @@ func (ts *TestServer) ClusterSettings() *cluster.Settings {
 	return ts.Cfg.Settings
 }
 
+// SettingsWatcher is part of the TestTenantInterface.
+func (ts *TestServer) SettingsWatcher() interface{} {
+	return ts.sqlServer.settingsWatcher
+}
+
 // Engines returns the TestServer's engines.
 func (ts *TestServer) Engines() []storage.Engine {
 	return ts.engines

pkg/testutils/serverutils/test_tenant_shim.go

@@ -99,6 +99,10 @@ type TestTenantInterface interface {
 	// this tenant.
 	ClusterSettings() *cluster.Settings
 
+	// SettingsWatcher returns the *settingswatcher.SettingsWatcher used by the
+	// tenant server.
+	SettingsWatcher() interface{}
+
 	// Stopper returns the stopper used by the tenant.
 	Stopper() *stop.Stopper