Merge pull request #151149 from celiala/backportrelease-25.3-150991

release-25.3: pgwire: limit concurrent updates to last_login_time

Author: rafiss

pkg/base/test_server_args.go

@@ -635,6 +635,10 @@ type TestSharedProcessTenantArgs struct {
 	// TenantID is the ID of the tenant to be created. If not set, an ID is
 	// assigned automatically.
 	TenantID roachpb.TenantID
+	// TenantReadOnly indicates if this tenant should be created as read-only
+	// (for testing PCR reader tenants). This field is used for testing purposes
+	// and overrides the tenant record check.
+	TenantReadOnly bool
 
 	Knobs TestingKnobs
 

pkg/server/config.go

@@ -487,6 +487,9 @@ type SQLConfig struct {
 	TenantID   roachpb.TenantID
 	TenantName roachpb.TenantName
 
+	// TenantReadOnly indicates if this tenant is read-only (PCR reader tenant).
+	TenantReadOnly bool
+
 	// If set, will to be called at server startup to obtain the tenant id and
 	// locality.
 	DelayedSetTenantID func(context.Context) (roachpb.TenantID, roachpb.Locality, error)
@@ -554,8 +557,9 @@ func MakeSQLConfig(
 	tenID roachpb.TenantID, tenName roachpb.TenantName, tempStorageCfg base.TempStorageConfig,
 ) SQLConfig {
 	sqlCfg := SQLConfig{
-		TenantID:   tenID,
-		TenantName: tenName,
+		TenantID:       tenID,
+		TenantName:     tenName,
+		TenantReadOnly: false, // Default to false, will be set during tenant initialization
 	}
 	sqlCfg.SetDefaults(tempStorageCfg)
 	return sqlCfg

pkg/server/server_controller_new_server.go

@@ -64,13 +64,18 @@ func (s *topLevelServer) newTenantServer(
 	portStartHint int,
 	testArgs base.TestSharedProcessTenantArgs,
 ) (onDemandServer, error) {
-	tenantID, err := s.getTenantID(ctx, tenantNameContainer.Get())
+	tenantID, tenantReadOnly, err := s.getTenantID(ctx, tenantNameContainer.Get())
 	if err != nil {
 		return nil, err
 	}
 
+	// Use test override for tenant read-only status if provided.
+	if testArgs.TenantReadOnly {
+		tenantReadOnly = true
+	}
+
 	baseCfg, sqlCfg, err := s.makeSharedProcessTenantConfig(ctx, tenantID, tenantNameContainer.Get(), portStartHint,
-		tenantStopper, testArgs.Settings)
+		tenantStopper, testArgs.Settings, tenantReadOnly)
 	if err != nil {
 		return nil, err
 	}
@@ -100,23 +105,27 @@ var ErrInvalidTenant error = errInvalidTenantMarker{}
 
 func (s *topLevelServer) getTenantID(
 	ctx context.Context, tenantName roachpb.TenantName,
-) (roachpb.TenantID, error) {
+) (roachpb.TenantID, bool, error) {
 	var rec *mtinfopb.TenantInfo
 	if err := s.sqlServer.internalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 		var err error
 		rec, err = sql.GetTenantRecordByName(ctx, s.cfg.Settings, txn, tenantName)
 		return err
 	}); err != nil {
-		return roachpb.TenantID{}, errors.Mark(err, ErrInvalidTenant)
+		return roachpb.TenantID{}, false, errors.Mark(err, ErrInvalidTenant)
 	}
 
 	tenantID, err := roachpb.MakeTenantID(rec.ID)
 	if err != nil {
-		return roachpb.TenantID{}, errors.Mark(
+		return roachpb.TenantID{}, false, errors.Mark(
 			errors.NewAssertionErrorWithWrappedErrf(err, "stored tenant ID %d does not convert to TenantID", rec.ID),
 			ErrInvalidTenant)
 	}
-	return tenantID, nil
+
+	// Check if tenant is read-only (PCR reader tenant).
+	readOnlyTenant := rec.ReadFromTenant != nil
+
+	return tenantID, readOnlyTenant, nil
 }
 
 // newTenantServerInternal instantiates a server for the given target
@@ -151,6 +160,7 @@ func (s *topLevelServer) makeSharedProcessTenantConfig(
 	portStartHint int,
 	stopper *stop.Stopper,
 	testSettings *cluster.Settings,
+	tenantReadOnly bool,
 ) (BaseConfig, SQLConfig, error) {
 	// Create a configuration for the new tenant.
 	parentCfg := s.cfg
@@ -167,7 +177,7 @@ func (s *topLevelServer) makeSharedProcessTenantConfig(
 	}
 
 	baseCfg, sqlCfg, err := makeSharedProcessTenantServerConfig(ctx, tenantID, tenantName, portStartHint, parentCfg,
-		localServerInfo, st, stopper, s.recorder)
+		localServerInfo, st, stopper, s.recorder, tenantReadOnly)
 	if err != nil {
 		return BaseConfig{}, SQLConfig{}, err
 	}
@@ -186,6 +196,7 @@ func makeSharedProcessTenantServerConfig(
 	st *cluster.Settings,
 	stopper *stop.Stopper,
 	nodeMetricsRecorder *status.MetricsRecorder,
+	tenantReadOnly bool,
 ) (baseCfg BaseConfig, sqlCfg SQLConfig, err error) {
 	tr := tracing.NewTracerWithOpt(ctx, tracing.WithClusterSettings(&st.SV))
 
@@ -333,6 +344,7 @@ func makeSharedProcessTenantServerConfig(
 	}
 
 	sqlCfg = MakeSQLConfig(tenantID, tenantName, tempStorageCfg)
+	sqlCfg.TenantReadOnly = tenantReadOnly
 	baseCfg.ExternalIODirConfig = kvServerCfg.BaseConfig.ExternalIODirConfig
 
 	baseCfg.ExternalIODir = kvServerCfg.BaseConfig.ExternalIODir

pkg/server/server_sql.go

@@ -1052,6 +1052,7 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 		RangeStatsFetcher:          rangeStatsFetcher,
 		NodeDescs:                  cfg.nodeDescs,
 		TenantCapabilitiesReader:   cfg.tenantCapabilitiesReader,
+		TenantReadOnly:             cfg.SQLConfig.TenantReadOnly,
 		CidrLookup:                 cfg.BaseConfig.CidrLookup,
 		LicenseEnforcer:            cfg.SQLConfig.LicenseEnforcer,
 	}

pkg/sql/exec_util.go

@@ -1677,6 +1677,9 @@ type ExecutorConfig struct {
 
 	TenantCapabilitiesReader SystemTenantOnly[tenantcapabilities.Reader]
 
+	// TenantReadOnly indicates if this tenant is read-only (PCR reader tenant).
+	TenantReadOnly bool
+
 	// VirtualClusterName contains the name of the virtual cluster
 	// (tenant).
 	VirtualClusterName roachpb.TenantName

pkg/sql/logictest/testdata/logic_test/user

@@ -240,9 +240,9 @@ onlyif config local-mixed-25.2
 statement ok
 DROP user testuser
 
-onlyif config local-mixed-25.2
 user testuser nodeidx=0 newsession
 
+onlyif config local-mixed-25.2
 statement error pq: password authentication failed for user testuser
 SHOW session_user
 
@@ -259,7 +259,6 @@ skipif config local-mixed-25.2
 statement ok
 DROP user testuser
 
-skipif config local-mixed-25.2
 user testuser nodeidx=0 newsession
 
 # The logictest suite currently doesn't work for external authentication methods
@@ -271,39 +270,39 @@ skipif config local-mixed-25.2
 statement error pq: user identity unknown for identity provider
 SHOW session_user
 
-subtest end
-
-subtest validate_estimated_last_login_time
-
 user root
 
 statement ok
 CREATE user IF NOT EXISTS testuser
 
-user testuser
+subtest end
+
+subtest validate_estimated_last_login_time
+
+user testuser2
 
 # Since the logictest framework does not connect with the user until a command
 # is executed, we need to run a command before checking estimated_last_login_time.
 query T
 SHOW session_user
 ----
-testuser
+testuser2
 
 user root
 
 # The estimated_last_login_time is not guaranteed to be populated synchronously,
 # so we poll until testuser's entry was updated with the last login time.
 skipif config local-mixed-25.2
 query I retry
-SELECT count(*) FROM system.users WHERE estimated_last_login_time IS NOT NULL AND username = 'testuser'
+SELECT count(*) FROM system.users WHERE estimated_last_login_time IS NOT NULL AND username = 'testuser2'
 ----
 1
 
 user root
 
 onlyif config local-mixed-25.2
 statement error pq: column "estimated_last_login_time" does not exist
-select estimated_last_login_time from [SHOW USERS] where username = 'testuser';
+select estimated_last_login_time from [SHOW USERS] where username = 'testuser2';
 
 subtest end
 

pkg/sql/pgwire/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "conn.go",
         "hba_conf.go",
         "ident_map_conf.go",
+        "last_login_updater.go",
         "pre_serve.go",
         "pre_serve_options.go",
         "role_mapper.go",
@@ -63,6 +64,7 @@ go_library(
         "//pkg/sql/sqltelemetry",
         "//pkg/sql/types",
         "//pkg/util",
+        "//pkg/util/cache",
         "//pkg/util/ctxlog",
         "//pkg/util/duration",
         "//pkg/util/envutil",
@@ -81,6 +83,7 @@ go_library(
         "//pkg/util/ring",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
+        "//pkg/util/syncutil/singleflight",
         "//pkg/util/system",
         "//pkg/util/timeofday",
         "//pkg/util/timeutil",

pkg/sql/pgwire/auth.go

@@ -90,14 +90,17 @@ type authOptions struct {
 // authentication and update c.sessionArgs with the authenticated user's name,
 // if different from the one given initially.
 func (c *conn) handleAuthentication(
-	ctx context.Context, ac AuthConn, authOpt authOptions, execCfg *sql.ExecutorConfig,
+	ctx context.Context, ac AuthConn, authOpt authOptions, server *Server,
 ) (connClose func(), _ error) {
 	if authOpt.testingSkipAuth {
 		return nil, nil
 	}
 	if authOpt.testingAuthHook != nil {
 		return nil, authOpt.testingAuthHook(ctx)
 	}
+	// Get execCfg from the server.
+	execCfg := server.execCfg
+
 	// To book-keep the authentication start time.
 	authStartTime := timeutil.Now()
 
@@ -278,28 +281,11 @@ func (c *conn) handleAuthentication(
 	duration := timeutil.Since(authStartTime).Nanoseconds()
 	c.publishConnLatencyMetric(duration, hbaEntry.Method.String())
 
-	c.populateLastLoginTime(ctx, execCfg, dbUser)
+	server.lastLoginUpdater.updateLastLoginTime(ctx, dbUser)
 
 	return connClose, nil
 }
 
-// populateLastLoginTime updates the last login time for the sql user
-// asynchronously. This not guaranteed to succeed and we log any errors obtained
-// from the update transaction to the DEV channel.
-func (c *conn) populateLastLoginTime(
-	ctx context.Context, execCfg *sql.ExecutorConfig, dbUser username.SQLUsername,
-) {
-	// Update last login time in async. This is done asynchronously to avoid
-	// blocking the connection.
-	if err := execCfg.Stopper.RunAsyncTask(ctx, "write_last_login_time", func(ctx context.Context) {
-		if err := sql.UpdateLastLoginTime(ctx, execCfg, dbUser.SQLIdentifier()); err != nil {
-			log.Warningf(ctx, "failed to update last login time for user %s: %v", dbUser, err)
-		}
-	}); err != nil {
-		log.Warningf(ctx, "failed to create async task to update last login time for user %s: %v", dbUser, err)
-	}
-}
-
 // publishConnLatencyMetric publishes the latency  of the connection
 // based on the authentication method.
 func (c *conn) publishConnLatencyMetric(duration int64, authMethod string) {

pkg/sql/pgwire/conn.go

@@ -151,7 +151,7 @@ func (c *conn) processCommands(
 	ctx context.Context,
 	authOpt authOptions,
 	ac AuthConn,
-	sqlServer *sql.Server,
+	server *Server,
 	reserved *mon.BoundAccount,
 	onDefaultIntSizeChange func(newSize int32),
 	sessionID clusterunique.ID,
@@ -175,7 +175,7 @@ func (c *conn) processCommands(
 		// network read on the connection's goroutine.
 		c.cancelConn()
 
-		pgwireKnobs := sqlServer.GetExecutorConfig().PGWireTestingKnobs
+		pgwireKnobs := server.SQLServer.GetExecutorConfig().PGWireTestingKnobs
 		if pgwireKnobs != nil && pgwireKnobs.CatchPanics {
 			if r := recover(); r != nil {
 				// Catch the panic and return it to the client as an error.
@@ -203,14 +203,14 @@ func (c *conn) processCommands(
 
 	// Authenticate the connection.
 	if connCloseAuthHandler, retErr = c.handleAuthentication(
-		ctx, ac, authOpt, sqlServer.GetExecutorConfig(),
+		ctx, ac, authOpt, server,
 	); retErr != nil {
 		// Auth failed or some other error.
 		return
 	}
 
 	var decrementConnectionCount func()
-	if decrementConnectionCount, retErr = sqlServer.IncrementConnectionCount(c.sessionArgs); retErr != nil {
+	if decrementConnectionCount, retErr = server.SQLServer.IncrementConnectionCount(c.sessionArgs); retErr != nil {
 		// This will return pgcode.TooManyConnections which is used by the sql proxy
 		// to skip failed auth throttle (as in this case the auth was fine but the
 		// error occurred before sending back auth ok msg)
@@ -224,7 +224,7 @@ func (c *conn) processCommands(
 	}
 
 	// Inform the client of the default session settings.
-	connHandler, retErr = c.sendInitialConnData(ctx, sqlServer, onDefaultIntSizeChange, sessionID)
+	connHandler, retErr = c.sendInitialConnData(ctx, server.SQLServer, onDefaultIntSizeChange, sessionID)
 	if retErr != nil {
 		return
 	}
@@ -250,7 +250,7 @@ func (c *conn) processCommands(
 
 	// Now actually process commands.
 	reservedOwned = false // We're about to pass ownership away.
-	retErr = sqlServer.ServeConn(
+	retErr = server.SQLServer.ServeConn(
 		ctx,
 		connHandler,
 		reserved,