server: Reject SQL Pods with an invalid tenant id.

This commit fixes an issue where attempting to start a SQL Pod
through cockroach mt start-sql command returns an error if the
tenant id has not been created before. Previously, attempting to
start a SQL Pod with a non existent tenant id would crash. Even
worse, it would poison the tenant id causing future
crdb_internal.create_sql_tenant calls to fail.
With this fix, the cockroach mt start-sql command returns an error
for non existent tenant ids and future calls to
crdb_internal.create_sql_tenant are successful.
Resolves: #64963
Release note (bug fix): The cockroach mt start-sql command with a
non existent tenant id returns an error. Previously, it would
crash and poison the tenant id for future usage.

pkg/base/test_server_args.go

@@ -268,4 +268,7 @@ type TestTenantArgs struct {
 	// automatically open a connection to the server. That's equivalent to running
 	// SET DATABASE=foo, which works even if the database doesn't (yet) exist.
 	UseDatabase string
+
+	// Skip check for tenant existence when running the test.
+	SkipTenantCheck bool
 }

pkg/ccl/serverccl/server_sql_test.go

@@ -205,3 +205,21 @@ func TestIdleExit(t *testing.T) {
 		t.Error("stop on idle didn't trigger")
 	}
 }
+
+func TestNonExistentTenant(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+
+	tc := serverutils.StartNewTestCluster(t, 1, base.TestClusterArgs{})
+	defer tc.Stopper().Stop(ctx)
+
+	_, err := tc.Server(0).StartTenant(ctx,
+		base.TestTenantArgs{
+			TenantID:        serverutils.TestTenantID(),
+			Existing:        true,
+			SkipTenantCheck: true,
+		})
+	require.Error(t, err)
+	require.Equal(t, "system DB uninitialized, check if tenant is non existent", err.Error())
+}

pkg/server/BUILD.bazel

@@ -108,6 +108,7 @@ go_library(
         "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/catalog/bootstrap",
+        "//pkg/sql/catalog/catalogkeys",
         "//pkg/sql/catalog/catconstants",
         "//pkg/sql/catalog/colinfo",
         "//pkg/sql/catalog/descpb",

pkg/server/server_sql.go

@@ -47,8 +47,10 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/server/settingswatcher"
 	"github.com/cockroachdb/cockroach/pkg/server/status"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkeys"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/hydratedtables"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/systemschema"
 	"github.com/cockroachdb/cockroach/pkg/sql/colexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/contention"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsql"
@@ -252,7 +254,6 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 			codec = keys.MakeSQLCodec(override)
 		}
 	}
-
 	// Create blob service for inter-node file sharing.
 	blobService, err := blobs.NewBlobService(cfg.Settings.ExternalIODir)
 	if err != nil {
@@ -746,6 +747,28 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 	}, nil
 }
 
+// Checks if tenant exists. This function does a very superficial check to see if the system db
+// has been bootstrapped for the tenant. This is not a complete check and is only sufficient
+// to be used in the dev environment.
+func maybeCheckTenantExists(ctx context.Context, codec keys.SQLCodec, db *kv.DB) error {
+	if codec.ForSystemTenant() {
+		// Skip check for system tenant and return early.
+		return nil
+	}
+	key := catalogkeys.NewDatabaseKey(systemschema.SystemDatabaseName).Key(codec)
+	result, err := db.Get(ctx, key)
+	if err != nil {
+		return err
+	}
+	if result.Value == nil || result.ValueInt() != keys.SystemDatabaseID {
+		return errors.New("system DB uninitialized, check if tenant is non existent")
+	}
+	// Tenant has been confirmed to be bootstrapped successfully
+	// as the system database, which is a part of the bootstrap data for
+	// a tenant keyspace, exists in the namespace table.
+	return nil
+}
+
 func (s *SQLServer) preStart(
 	ctx context.Context,
 	stopper *stop.Stopper,
@@ -764,6 +787,13 @@ func (s *SQLServer) preStart(
 			return err
 		}
 	}
+	// Confirm tenant exists prior to initialization. This is a sanity
+	// check for the dev environment to ensure that a tenant has been
+	// successfully created before attempting to initialize a SQL
+	// server for it.
+	if err := maybeCheckTenantExists(ctx, s.execCfg.Codec, s.execCfg.DB); err != nil {
+		return err
+	}
 	s.connManager = connManager
 	s.pgL = pgL
 	s.execCfg.GCJobNotifier.Start(ctx)

pkg/server/testserver.go

@@ -716,18 +716,20 @@ func (ts *TestServer) StartTenant(
 		}
 	}
 
-	rowCount, err := ts.InternalExecutor().(*sql.InternalExecutor).Exec(
-		ctx, "testserver-check-tenant-active", nil,
-		"SELECT 1 FROM system.tenants WHERE id=$1 AND active=true",
-		params.TenantID.ToUint64(),
-	)
-	if err != nil {
-		return nil, err
-	}
-	if rowCount == 0 {
-		return nil, errors.New("not found")
-	}
+	if !params.SkipTenantCheck {
+		rowCount, err := ts.InternalExecutor().(*sql.InternalExecutor).Exec(
+			ctx, "testserver-check-tenant-active", nil,
+			"SELECT 1 FROM system.tenants WHERE id=$1 AND active=true",
+			params.TenantID.ToUint64(),
+		)
 
+		if err != nil {
+			return nil, err
+		}
+		if rowCount == 0 {
+			return nil, errors.New("not found")
+		}
+	}
 	st := params.Settings
 	if st == nil {
 		st = cluster.MakeTestingClusterSettings()