[v14] Add RunWhileLockedRetryInterval setting to TestAuthServer

lib/auth/helpers.go

@@ -88,6 +88,9 @@ type TestAuthServerConfig struct {
 	Embedder embedding.Embedder
 	// CacheEnabled enables the primary auth server cache.
 	CacheEnabled bool
+	// RunWhileLockedRetryInterval is the interval to retry the run while locked
+	// operation.
+	RunWhileLockedRetryInterval time.Duration
 }
 
 // CheckAndSetDefaults checks and sets defaults
@@ -276,6 +279,11 @@ func NewTestAuthServer(cfg TestAuthServerConfig) (*TestAuthServer, error) {
 		return nil, trace.Wrap(err)
 	}
 
+	accessLists, err := local.NewAccessListService(srv.Backend, cfg.Clock, local.WithRunWhileLockedRetryInterval(cfg.RunWhileLockedRetryInterval))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
 	srv.AuthServer, err = NewServer(&InitConfig{
 		Backend:                srv.Backend,
 		Authority:              authority.NewWithClock(cfg.Clock),
@@ -293,6 +301,7 @@ func NewTestAuthServer(cfg TestAuthServerConfig) (*TestAuthServer, error) {
 		},
 		EmbeddingRetriever: ai.NewSimpleRetriever(),
 		HostUUID:           uuid.New().String(),
+		AccessLists:        accessLists,
 	},
 		WithClock(cfg.Clock),
 		WithEmbedder(cfg.Embedder),

lib/services/local/access_list.go

@@ -70,38 +70,45 @@ type AccessListService struct {
 var _ services.AccessLists = (*AccessListService)(nil)
 
 // NewAccessListService creates a new AccessListService.
-func NewAccessListService(backend backend.Backend, clock clockwork.Clock) (*AccessListService, error) {
+func NewAccessListService(backend backend.Backend, clock clockwork.Clock, opts ...ServiceOption) (*AccessListService, error) {
+	var opt serviceOptions
+	for _, o := range opts {
+		o(&opt)
+	}
 	service, err := generic.NewService(&generic.ServiceConfig[*accesslist.AccessList]{
-		Backend:       backend,
-		PageLimit:     accessListMaxPageSize,
-		ResourceKind:  types.KindAccessList,
-		BackendPrefix: accessListPrefix,
-		MarshalFunc:   services.MarshalAccessList,
-		UnmarshalFunc: services.UnmarshalAccessList,
+		Backend:                     backend,
+		PageLimit:                   accessListMaxPageSize,
+		ResourceKind:                types.KindAccessList,
+		BackendPrefix:               accessListPrefix,
+		MarshalFunc:                 services.MarshalAccessList,
+		UnmarshalFunc:               services.UnmarshalAccessList,
+		RunWhileLockedRetryInterval: opt.runWhileLockedRetryInterval,
 	})
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
 
 	memberService, err := generic.NewService(&generic.ServiceConfig[*accesslist.AccessListMember]{
-		Backend:       backend,
-		PageLimit:     accessListMemberMaxPageSize,
-		ResourceKind:  types.KindAccessListMember,
-		BackendPrefix: accessListMemberPrefix,
-		MarshalFunc:   services.MarshalAccessListMember,
-		UnmarshalFunc: services.UnmarshalAccessListMember,
+		Backend:                     backend,
+		PageLimit:                   accessListMemberMaxPageSize,
+		ResourceKind:                types.KindAccessListMember,
+		BackendPrefix:               accessListMemberPrefix,
+		MarshalFunc:                 services.MarshalAccessListMember,
+		UnmarshalFunc:               services.UnmarshalAccessListMember,
+		RunWhileLockedRetryInterval: opt.runWhileLockedRetryInterval,
 	})
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
 
 	reviewService, err := generic.NewService(&generic.ServiceConfig[*accesslist.Review]{
-		Backend:       backend,
-		PageLimit:     accessListReviewMaxPageSize,
-		ResourceKind:  types.KindAccessListReview,
-		BackendPrefix: accessListReviewPrefix,
-		MarshalFunc:   services.MarshalAccessListReview,
-		UnmarshalFunc: services.UnmarshalAccessListReview,
+		Backend:                     backend,
+		PageLimit:                   accessListReviewMaxPageSize,
+		ResourceKind:                types.KindAccessListReview,
+		BackendPrefix:               accessListReviewPrefix,
+		MarshalFunc:                 services.MarshalAccessListReview,
+		UnmarshalFunc:               services.UnmarshalAccessListReview,
+		RunWhileLockedRetryInterval: opt.runWhileLockedRetryInterval,
 	})
 	if err != nil {
 		return nil, trace.Wrap(err)

lib/services/local/generic/generic.go

@@ -49,6 +49,10 @@ type ServiceConfig[T Resource] struct {
 	BackendPrefix string
 	MarshalFunc   MarshalFunc[T]
 	UnmarshalFunc UnmarshalFunc[T]
+	// RunWhileLockedRetryInterval is the interval to retry the RunWhileLocked function.
+	// If set to 0, the default interval of 250ms will be used.
+	// WARNING: If set to a negative value, the RunWhileLocked function will retry immediately.
+	RunWhileLockedRetryInterval time.Duration
 }
 
 func (c *ServiceConfig[T]) CheckAndSetDefaults() error {
@@ -78,12 +82,13 @@ func (c *ServiceConfig[T]) CheckAndSetDefaults() error {
 
 // Service is a generic service for interacting with resources in the backend.
 type Service[T Resource] struct {
-	backend       backend.Backend
-	resourceKind  string
-	pageLimit     uint
-	backendPrefix string
-	marshalFunc   MarshalFunc[T]
-	unmarshalFunc UnmarshalFunc[T]
+	backend                     backend.Backend
+	resourceKind                string
+	pageLimit                   uint
+	backendPrefix               string
+	marshalFunc                 MarshalFunc[T]
+	unmarshalFunc               UnmarshalFunc[T]
+	runWhileLockedRetryInterval time.Duration
 }
 
 // NewService will return a new generic service with the given config. This will
@@ -94,12 +99,13 @@ func NewService[T Resource](cfg *ServiceConfig[T]) (*Service[T], error) {
 	}
 
 	return &Service[T]{
-		backend:       cfg.Backend,
-		resourceKind:  cfg.ResourceKind,
-		pageLimit:     cfg.PageLimit,
-		backendPrefix: cfg.BackendPrefix,
-		marshalFunc:   cfg.MarshalFunc,
-		unmarshalFunc: cfg.UnmarshalFunc,
+		backend:                     cfg.Backend,
+		resourceKind:                cfg.ResourceKind,
+		pageLimit:                   cfg.PageLimit,
+		backendPrefix:               cfg.BackendPrefix,
+		marshalFunc:                 cfg.MarshalFunc,
+		unmarshalFunc:               cfg.UnmarshalFunc,
+		runWhileLockedRetryInterval: cfg.RunWhileLockedRetryInterval,
 	}, nil
 }
 
@@ -375,9 +381,10 @@ func (s *Service[T]) RunWhileLocked(ctx context.Context, lockName string, ttl ti
 	return trace.Wrap(backend.RunWhileLocked(ctx,
 		backend.RunWhileLockedConfig{
 			LockConfiguration: backend.LockConfiguration{
-				Backend:  s.backend,
-				LockName: lockName,
-				TTL:      ttl,
+				Backend:       s.backend,
+				LockName:      lockName,
+				TTL:           ttl,
+				RetryInterval: s.runWhileLockedRetryInterval,
 			},
 		}, func(ctx context.Context) error {
 			return fn(ctx, s.backend)

lib/services/local/options.go

@@ -0,0 +1,36 @@
+/*
+ * Teleport
+ * Copyright (C) 2023  Gravitational, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package local
+
+import "time"
+
+// ServiceOption is a functional option for configuring the service.
+// TODO(tigrato): Add support for other services besides the access list service.
+type ServiceOption func(*serviceOptions)
+
+type serviceOptions struct {
+	runWhileLockedRetryInterval time.Duration
+}
+
+// WithRunWhileLockedRetryInterval sets the retry interval for the RunWhileLocked function.
+func WithRunWhileLockedRetryInterval(interval time.Duration) ServiceOption {
+	return func(o *serviceOptions) {
+		o.runWhileLockedRetryInterval = interval
+	}
+}