Split grace period into multiple config values (#2578) (#2603)

Co-authored-by: Todd <github@quaddi.com>

internal/cmd/base/servers.go

@@ -26,7 +26,6 @@ import (
 	berrors "github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/observability/event"
-	"github.com/hashicorp/boundary/internal/server"
 	"github.com/hashicorp/boundary/internal/types/scope"
 	kms_plugin_assets "github.com/hashicorp/boundary/plugins/kms"
 	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
@@ -48,21 +47,6 @@ import (
 )
 
 const (
-	// defaultStatusGracePeriod is the default status grace period, or the period
-	// of time that we will go without a status report before we start
-	// disconnecting and marking connections as closed. This is tied to the
-	// server default liveness setting, a related value. See the server package
-	// for more details.
-	defaultStatusGracePeriod = server.DefaultLiveness
-
-	// statusGracePeriodEnvVar is the environment variable that can be used to
-	// configure the status grace period. This setting is provided in seconds,
-	// and can never be lower than the default status grace period defined above.
-	//
-	// TODO: This value is temporary, it will be removed once we have a better
-	// story/direction on attributes and system defaults.
-	statusGracePeriodEnvVar = "BOUNDARY_STATUS_GRACE_PERIOD"
-
 	// File name to use for storing workerAuth requests
 	WorkerAuthReqFile = "auth_request_token"
 )
@@ -144,11 +128,6 @@ type Server struct {
 	DevDatabaseCleanupFunc func() error
 
 	Database *db.DB
-
-	// StatusGracePeriodDuration represents the period of time (as a
-	// duration) that the controller will wait before marking
-	// connections from a disconnected worker as invalid.
-	StatusGracePeriodDuration time.Duration
 }
 
 // NewServer creates a new Server.
@@ -807,48 +786,3 @@ func MakeSighupCh() chan struct{} {
 	}()
 	return resultCh
 }
-
-// SetStatusGracePeriodDuration sets the value for
-// StatusGracePeriodDuration.
-//
-// The grace period is the length of time we allow connections to run
-// on a worker in the event of an error sending status updates. The
-// period is defined the length of time since the last successful
-// update.
-//
-// The setting is derived from one of the following, in order:
-//
-//   - Via the supplied value if non-zero.
-//   - BOUNDARY_STATUS_GRACE_PERIOD, if defined, can be set to an
-//     integer value to define the setting.
-//   - If either of these is missing, the default is used. See the
-//     defaultStatusGracePeriod value for the default value.
-//
-// The minimum setting for this value is the default setting. Values
-// below this will be reset to the default.
-func (s *Server) SetStatusGracePeriodDuration(value time.Duration) {
-	const op = "base.(Server).SetStatusGracePeriodDuration"
-	ctx := context.TODO()
-	var result time.Duration
-	switch {
-	case value > 0:
-		result = value
-	case os.Getenv(statusGracePeriodEnvVar) != "":
-		// TODO: See the description of the constant for more details on
-		// this env var
-		v := os.Getenv(statusGracePeriodEnvVar)
-		n, err := strconv.Atoi(v)
-		if err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("could not read status grace period setting", "envvar", statusGracePeriodEnvVar, "value", v))
-			break
-		}
-
-		result = time.Second * time.Duration(n)
-	}
-
-	if result < defaultStatusGracePeriod {
-		result = defaultStatusGracePeriod
-	}
-
-	s.StatusGracePeriodDuration = result
-}

internal/cmd/commands/dev/dev.go

@@ -565,10 +565,6 @@ func (c *Command) Run(args []string) int {
 		return base.CommandUserError
 	}
 
-	// Initialize status grace period (0 denotes using env or default
-	// here)
-	c.SetStatusGracePeriodDuration(0)
-
 	base.StartMemProfiler(c.Context)
 
 	if err := c.SetupEventing(

internal/cmd/commands/server/server.go

@@ -173,10 +173,6 @@ func (c *Command) Run(args []string) int {
 		return base.CommandUserError
 	}
 
-	// Initialize status grace period (0 denotes using env or default
-	// here)
-	c.SetStatusGracePeriodDuration(0)
-
 	base.StartMemProfiler(c.Context)
 
 	// Note: the checks directly after this must remain where they are because

internal/cmd/config/config.go

@@ -19,6 +19,7 @@ import (
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/util"
 	kms_plugin_assets "github.com/hashicorp/boundary/plugins/kms"
 	"github.com/hashicorp/boundary/sdk/wrapper"
 	"github.com/hashicorp/go-hclog"
@@ -145,26 +146,45 @@ type Controller struct {
 	Scheduler         *Scheduler `hcl:"scheduler"`
 
 	// AuthTokenTimeToLive is the total valid lifetime of a token denoted by time.Duration
-	AuthTokenTimeToLive         any `hcl:"auth_token_time_to_live"`
-	AuthTokenTimeToLiveDuration time.Duration
+	AuthTokenTimeToLive         any           `hcl:"auth_token_time_to_live"`
+	AuthTokenTimeToLiveDuration time.Duration `hcl:"-"`
 
 	// AuthTokenTimeToStale is the total time a token can go unused before becoming invalid
 	// denoted by time.Duration
-	AuthTokenTimeToStale         any `hcl:"auth_token_time_to_stale"`
-	AuthTokenTimeToStaleDuration time.Duration
+	AuthTokenTimeToStale         any           `hcl:"auth_token_time_to_stale"`
+	AuthTokenTimeToStaleDuration time.Duration `hcl:"-"`
 
 	// GracefulShutdownWait is the amount of time that we'll wait before actually
 	// starting the Controller shutdown. This allows the health endpoint to
 	// return a status code to indicate that the instance is shutting down.
-	GracefulShutdownWait         any `hcl:"graceful_shutdown_wait_duration"`
-	GracefulShutdownWaitDuration time.Duration
+	GracefulShutdownWait         any           `hcl:"graceful_shutdown_wait_duration"`
+	GracefulShutdownWaitDuration time.Duration `hcl:"-"`
 
-	// StatusGracePeriod represents the period of time (as a duration) that the
-	// controller will wait before marking connections from a disconnected worker
-	// as invalid.
+	// WorkerStatusGracePeriod represents the period of time (as a duration)
+	// that the controller will wait before deciding a worker is disconnected
+	// and marking connections from it as canceling
+	//
+	// TODO: This isn't documented (on purpose) because the right place for this
+	// is central configuration so you can't drift across controllers, but we
+	// don't have that yet.
+	WorkerStatusGracePeriod         interface{}   `hcl:"worker_status_grace_period"`
+	WorkerStatusGracePeriodDuration time.Duration `hcl:"-"`
+
+	// LivenessTimeToStale represents the period of time (as a duration) after
+	// which it will consider other controllers to be no longer accessible,
+	// based on time since their last status update in the database
+	//
+	// TODO: This isn't documented (on purpose) because the right place for this
+	// is central configuration so you can't drift across controllers, but we
+	// don't have that yet.
+	LivenessTimeToStale         interface{}   `hcl:"liveness_time_to_stale"`
+	LivenessTimeToStaleDuration time.Duration `hcl:"-"`
+
+	// SchedulerRunJobInterval is the time interval between waking up the
+	// scheduler to run pending jobs.
 	//
 	// TODO: This field is currently internal.
-	StatusGracePeriodDuration time.Duration `hcl:"-"`
+	SchedulerRunJobInterval time.Duration `hcl:"-"`
 }
 
 func (c *Controller) InitNameIfEmpty() error {
@@ -206,12 +226,22 @@ type Worker struct {
 	Tags    map[string][]string `hcl:"-"`
 	TagsRaw any                 `hcl:"tags"`
 
-	// StatusGracePeriod represents the period of time (as a duration) that the
-	// worker will wait before disconnecting connections if it cannot make a
-	// status report to a controller.
+	// StatusCallTimeout represents the period of time (as a duration) that
+	// the worker will allow a status RPC call to attempt to finish before
+	// canceling it to try again.
 	//
-	// TODO: This field is currently internal.
-	StatusGracePeriodDuration time.Duration `hcl:"-"`
+	// TODO: This is currently not documented and considered internal.
+	StatusCallTimeout         interface{}   `hcl:"status_call_timeout"`
+	StatusCallTimeoutDuration time.Duration `hcl:"-"`
+
+	// SuccessfulStatusGracePeriod represents the period of time (as a duration)
+	// that the worker will wait before disconnecting connections if it cannot
+	// successfully complete a status report to a controller. This cannot be
+	// less than StatusCallTimeout.
+	//
+	// TODO: This is currently not documented and considered internal.
+	SuccessfulStatusGracePeriod         interface{}   `hcl:"successful_status_grace_period"`
+	SuccessfulStatusGracePeriodDuration time.Duration `hcl:"-"`
 
 	// AuthStoragePath represents the location a worker stores its node credentials, if set
 	AuthStoragePath string `hcl:"auth_storage_path"`
@@ -498,6 +528,36 @@ func Parse(d string) (*Config, error) {
 			}
 		}
 
+		workerStatusGracePeriod := result.Controller.WorkerStatusGracePeriod
+		if util.IsNil(workerStatusGracePeriod) {
+			workerStatusGracePeriod = os.Getenv("BOUNDARY_CONTROLLER_WORKER_STATUS_GRACE_PERIOD")
+		}
+		if workerStatusGracePeriod != nil {
+			t, err := parseutil.ParseDurationSecond(workerStatusGracePeriod)
+			if err != nil {
+				return result, err
+			}
+			result.Controller.WorkerStatusGracePeriodDuration = t
+		}
+		if result.Controller.WorkerStatusGracePeriodDuration < 0 {
+			return nil, errors.New("Controller worker status grace period value is negative")
+		}
+
+		livenessTimeToStale := result.Controller.LivenessTimeToStale
+		if util.IsNil(livenessTimeToStale) {
+			livenessTimeToStale = os.Getenv("BOUNDARY_CONTROLLER_LIVENESS_TIME_TO_STALE")
+		}
+		if livenessTimeToStale != nil {
+			t, err := parseutil.ParseDurationSecond(livenessTimeToStale)
+			if err != nil {
+				return result, err
+			}
+			result.Controller.LivenessTimeToStaleDuration = t
+		}
+		if result.Controller.LivenessTimeToStaleDuration < 0 {
+			return nil, errors.New("Controller liveness time to stale value is negative")
+		}
+
 		if result.Controller.Database != nil {
 			if result.Controller.Database.MaxOpenConnectionsRaw != nil {
 				switch t := result.Controller.Database.MaxOpenConnectionsRaw.(type) {
@@ -583,6 +643,47 @@ func Parse(d string) (*Config, error) {
 			return nil, fmt.Errorf("Error parsing worker activation token: %w", err)
 		}
 
+		statusCallTimeoutDuration := result.Worker.StatusCallTimeout
+		if util.IsNil(statusCallTimeoutDuration) {
+			statusCallTimeoutDuration = os.Getenv("BOUNDARY_WORKER_STATUS_CALL_TIMEOUT")
+		}
+		if statusCallTimeoutDuration != nil {
+			t, err := parseutil.ParseDurationSecond(statusCallTimeoutDuration)
+			if err != nil {
+				return result, err
+			}
+			result.Worker.StatusCallTimeoutDuration = t
+		}
+		if result.Worker.StatusCallTimeoutDuration < 0 {
+			return nil, errors.New("Status call timeout value is negative")
+		}
+
+		successfulStatusGracePeriod := result.Worker.SuccessfulStatusGracePeriod
+		if util.IsNil(successfulStatusGracePeriod) {
+			successfulStatusGracePeriod = os.Getenv("BOUNDARY_WORKER_SUCCESSFUL_STATUS_GRACE_PERIOD")
+		}
+		if successfulStatusGracePeriod != nil {
+			t, err := parseutil.ParseDurationSecond(successfulStatusGracePeriod)
+			if err != nil {
+				return result, err
+			}
+			result.Worker.SuccessfulStatusGracePeriodDuration = t
+		}
+		if result.Worker.SuccessfulStatusGracePeriodDuration < 0 {
+			return nil, errors.New("Successful status grace period value is negative")
+		}
+
+		switch {
+		case result.Worker.StatusCallTimeoutDuration == 0 && result.Worker.SuccessfulStatusGracePeriodDuration == 0:
+			// Nothing
+		case result.Worker.StatusCallTimeoutDuration != 0 && result.Worker.SuccessfulStatusGracePeriodDuration != 0:
+			if result.Worker.StatusCallTimeoutDuration > result.Worker.SuccessfulStatusGracePeriodDuration {
+				return nil, fmt.Errorf("Worker setting for status call timeout duration must be less than or equal to successful status grace period duration")
+			}
+		default:
+			return nil, fmt.Errorf("Worker settings for status call timeout duration and successful status grace period duration must either both be set or both be empty")
+		}
+
 		if result.Worker.TagsRaw != nil {
 			switch t := result.Worker.TagsRaw.(type) {
 			// We allow `tags` to be a simple string containing a URL with schema.

internal/cmd/config/config_load_test.go

@@ -347,17 +347,18 @@ func TestLoad(t *testing.T) {
 					ClusterName:                  "",
 				},
 				Worker: &config.Worker{
-					Name:                               "worker0",
-					Description:                        "A worker",
-					PublicAddr:                         "",
-					InitialUpstreams:                   []string{"boundary:9201"},
-					InitialUpstreamsRaw:                []any{"boundary:9201"},
-					ControllersRaw:                     nil,
-					Tags:                               nil,
-					TagsRaw:                            nil,
-					StatusGracePeriodDuration:          0,
-					AuthStoragePath:                    "",
-					ControllerGeneratedActivationToken: "",
+					Name:                                "worker0",
+					Description:                         "A worker",
+					PublicAddr:                          "",
+					InitialUpstreams:                    []string{"boundary:9201"},
+					InitialUpstreamsRaw:                 []any{"boundary:9201"},
+					ControllersRaw:                      nil,
+					Tags:                                nil,
+					TagsRaw:                             nil,
+					SuccessfulStatusGracePeriodDuration: 0,
+					StatusCallTimeoutDuration:           0,
+					AuthStoragePath:                     "",
+					ControllerGeneratedActivationToken:  "",
 				},
 				Controller: &config.Controller{
 					Name:        "controller0",
@@ -373,15 +374,16 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:            "",
-					Scheduler:                    nil,
-					AuthTokenTimeToLive:          nil,
-					AuthTokenTimeToLiveDuration:  0,
-					AuthTokenTimeToStale:         nil,
-					AuthTokenTimeToStaleDuration: 0,
-					GracefulShutdownWait:         nil,
-					GracefulShutdownWaitDuration: 0,
-					StatusGracePeriodDuration:    0,
+					PublicClusterAddr:               "",
+					Scheduler:                       nil,
+					AuthTokenTimeToLive:             nil,
+					AuthTokenTimeToLiveDuration:     0,
+					AuthTokenTimeToStale:            nil,
+					AuthTokenTimeToStaleDuration:    0,
+					GracefulShutdownWait:            nil,
+					GracefulShutdownWaitDuration:    0,
+					WorkerStatusGracePeriodDuration: 0,
+					LivenessTimeToStaleDuration:     0,
 				},
 				DevController:           false,
 				DevUiPassthroughDir:     "",
@@ -751,17 +753,18 @@ func TestLoad(t *testing.T) {
 					ClusterName:                  "",
 				},
 				Worker: &config.Worker{
-					Name:                               "worker0",
-					Description:                        "A worker",
-					PublicAddr:                         "",
-					InitialUpstreams:                   []string{"boundary:9201"},
-					InitialUpstreamsRaw:                []any{"boundary:9201"},
-					ControllersRaw:                     nil,
-					Tags:                               nil,
-					TagsRaw:                            nil,
-					StatusGracePeriodDuration:          0,
-					AuthStoragePath:                    "",
-					ControllerGeneratedActivationToken: "",
+					Name:                                "worker0",
+					Description:                         "A worker",
+					PublicAddr:                          "",
+					InitialUpstreams:                    []string{"boundary:9201"},
+					InitialUpstreamsRaw:                 []any{"boundary:9201"},
+					ControllersRaw:                      nil,
+					Tags:                                nil,
+					TagsRaw:                             nil,
+					SuccessfulStatusGracePeriodDuration: 0,
+					StatusCallTimeoutDuration:           0,
+					AuthStoragePath:                     "",
+					ControllerGeneratedActivationToken:  "",
 				},
 				Controller: &config.Controller{
 					Name:        "controller0",
@@ -777,15 +780,16 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:            "",
-					Scheduler:                    nil,
-					AuthTokenTimeToLive:          nil,
-					AuthTokenTimeToLiveDuration:  0,
-					AuthTokenTimeToStale:         nil,
-					AuthTokenTimeToStaleDuration: 0,
-					GracefulShutdownWait:         nil,
-					GracefulShutdownWaitDuration: 0,
-					StatusGracePeriodDuration:    0,
+					PublicClusterAddr:               "",
+					Scheduler:                       nil,
+					AuthTokenTimeToLive:             nil,
+					AuthTokenTimeToLiveDuration:     0,
+					AuthTokenTimeToStale:            nil,
+					AuthTokenTimeToStaleDuration:    0,
+					GracefulShutdownWait:            nil,
+					GracefulShutdownWaitDuration:    0,
+					WorkerStatusGracePeriodDuration: 0,
+					LivenessTimeToStaleDuration:     0,
 				},
 				DevController:           false,
 				DevUiPassthroughDir:     "",
@@ -1155,17 +1159,18 @@ func TestLoad(t *testing.T) {
 					ClusterName:                  "",
 				},
 				Worker: &config.Worker{
-					Name:                               "worker0",
-					Description:                        "A worker",
-					PublicAddr:                         "",
-					InitialUpstreams:                   []string{"boundary:9201"},
-					InitialUpstreamsRaw:                []any{"boundary:9201"},
-					ControllersRaw:                     nil,
-					Tags:                               nil,
-					TagsRaw:                            nil,
-					StatusGracePeriodDuration:          0,
-					AuthStoragePath:                    "",
-					ControllerGeneratedActivationToken: "",
+					Name:                                "worker0",
+					Description:                         "A worker",
+					PublicAddr:                          "",
+					InitialUpstreams:                    []string{"boundary:9201"},
+					InitialUpstreamsRaw:                 []any{"boundary:9201"},
+					ControllersRaw:                      nil,
+					Tags:                                nil,
+					TagsRaw:                             nil,
+					SuccessfulStatusGracePeriodDuration: 0,
+					StatusCallTimeoutDuration:           0,
+					AuthStoragePath:                     "",
+					ControllerGeneratedActivationToken:  "",
 				},
 				Controller: &config.Controller{
 					Name:        "controller0",
@@ -1181,15 +1186,16 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:            "",
-					Scheduler:                    nil,
-					AuthTokenTimeToLive:          nil,
-					AuthTokenTimeToLiveDuration:  0,
-					AuthTokenTimeToStale:         nil,
-					AuthTokenTimeToStaleDuration: 0,
-					GracefulShutdownWait:         nil,
-					GracefulShutdownWaitDuration: 0,
-					StatusGracePeriodDuration:    0,
+					PublicClusterAddr:               "",
+					Scheduler:                       nil,
+					AuthTokenTimeToLive:             nil,
+					AuthTokenTimeToLiveDuration:     0,
+					AuthTokenTimeToStale:            nil,
+					AuthTokenTimeToStaleDuration:    0,
+					GracefulShutdownWait:            nil,
+					GracefulShutdownWaitDuration:    0,
+					WorkerStatusGracePeriodDuration: 0,
+					LivenessTimeToStaleDuration:     0,
 				},
 				DevController:           false,
 				DevUiPassthroughDir:     "",
@@ -1559,17 +1565,18 @@ func TestLoad(t *testing.T) {
 					ClusterName:                  "",
 				},
 				Worker: &config.Worker{
-					Name:                               "worker0",
-					Description:                        "A worker",
-					PublicAddr:                         "",
-					InitialUpstreams:                   []string{"boundary:9201"},
-					InitialUpstreamsRaw:                []any{"boundary:9201"},
-					ControllersRaw:                     nil,
-					Tags:                               nil,
-					TagsRaw:                            nil,
-					StatusGracePeriodDuration:          0,
-					AuthStoragePath:                    "",
-					ControllerGeneratedActivationToken: "",
+					Name:                                "worker0",
+					Description:                         "A worker",
+					PublicAddr:                          "",
+					InitialUpstreams:                    []string{"boundary:9201"},
+					InitialUpstreamsRaw:                 []any{"boundary:9201"},
+					ControllersRaw:                      nil,
+					Tags:                                nil,
+					TagsRaw:                             nil,
+					SuccessfulStatusGracePeriodDuration: 0,
+					StatusCallTimeoutDuration:           0,
+					AuthStoragePath:                     "",
+					ControllerGeneratedActivationToken:  "",
 				},
 				Controller: &config.Controller{
 					Name:        "controller0",
@@ -1585,15 +1592,16 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:            "",
-					Scheduler:                    nil,
-					AuthTokenTimeToLive:          nil,
-					AuthTokenTimeToLiveDuration:  0,
-					AuthTokenTimeToStale:         nil,
-					AuthTokenTimeToStaleDuration: 0,
-					GracefulShutdownWait:         nil,
-					GracefulShutdownWaitDuration: 0,
-					StatusGracePeriodDuration:    0,
+					PublicClusterAddr:               "",
+					Scheduler:                       nil,
+					AuthTokenTimeToLive:             nil,
+					AuthTokenTimeToLiveDuration:     0,
+					AuthTokenTimeToStale:            nil,
+					AuthTokenTimeToStaleDuration:    0,
+					GracefulShutdownWait:            nil,
+					GracefulShutdownWaitDuration:    0,
+					WorkerStatusGracePeriodDuration: 0,
+					LivenessTimeToStaleDuration:     0,
 				},
 				DevController:           false,
 				DevUiPassthroughDir:     "",
@@ -1669,15 +1677,16 @@ func TestLoad(t *testing.T) {
 						ConnMaxIdleTimeDuration:   nil,
 						SkipSharedLockAcquisition: false,
 					},
-					PublicClusterAddr:            "",
-					Scheduler:                    nil,
-					AuthTokenTimeToLive:          nil,
-					AuthTokenTimeToLiveDuration:  0,
-					AuthTokenTimeToStale:         nil,
-					AuthTokenTimeToStaleDuration: 0,
-					GracefulShutdownWait:         nil,
-					GracefulShutdownWaitDuration: 0,
-					StatusGracePeriodDuration:    0,
+					PublicClusterAddr:               "",
+					Scheduler:                       nil,
+					AuthTokenTimeToLive:             nil,
+					AuthTokenTimeToLiveDuration:     0,
+					AuthTokenTimeToStale:            nil,
+					AuthTokenTimeToStaleDuration:    0,
+					GracefulShutdownWait:            nil,
+					GracefulShutdownWaitDuration:    0,
+					WorkerStatusGracePeriodDuration: 0,
+					LivenessTimeToStaleDuration:     0,
 				},
 				DevController:           false,
 				DevUiPassthroughDir:     "",

internal/daemon/cluster/handlers/worker_service.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/daemon/controller/common"
@@ -28,12 +29,13 @@ type workerServiceServer struct {
 	pbs.UnsafeServerCoordinationServiceServer
 	pbs.UnsafeSessionServiceServer
 
-	serversRepoFn    common.ServersRepoFactory
-	workerAuthRepoFn common.WorkerAuthRepoStorageFactory
-	sessionRepoFn    session.RepositoryFactory
-	connectionRepoFn common.ConnectionRepoFactory
-	updateTimes      *sync.Map
-	kms              *kms.Kms
+	serversRepoFn       common.ServersRepoFactory
+	workerAuthRepoFn    common.WorkerAuthRepoStorageFactory
+	sessionRepoFn       session.RepositoryFactory
+	connectionRepoFn    common.ConnectionRepoFactory
+	updateTimes         *sync.Map
+	kms                 *kms.Kms
+	livenessTimeToStale *atomic.Int64
 }
 
 var (
@@ -48,14 +50,16 @@ func NewWorkerServiceServer(
 	connectionRepoFn common.ConnectionRepoFactory,
 	updateTimes *sync.Map,
 	kms *kms.Kms,
+	livenessTimeToStale *atomic.Int64,
 ) *workerServiceServer {
 	return &workerServiceServer{
-		serversRepoFn:    serversRepoFn,
-		workerAuthRepoFn: workerAuthRepoFn,
-		sessionRepoFn:    sessionRepoFn,
-		connectionRepoFn: connectionRepoFn,
-		updateTimes:      updateTimes,
-		kms:              kms,
+		serversRepoFn:       serversRepoFn,
+		workerAuthRepoFn:    workerAuthRepoFn,
+		sessionRepoFn:       sessionRepoFn,
+		connectionRepoFn:    connectionRepoFn,
+		updateTimes:         updateTimes,
+		kms:                 kms,
+		livenessTimeToStale: livenessTimeToStale,
 	}
 }
 
@@ -124,7 +128,7 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 		event.WriteError(ctx, op, err, event.WithInfoMsg("error storing worker status"))
 		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error storing worker status: %v", err)
 	}
-	controllers, err := serverRepo.ListControllers(ctx)
+	controllers, err := serverRepo.ListControllers(ctx, server.WithLiveness(time.Duration(ws.livenessTimeToStale.Load())))
 	if err != nil {
 		event.WriteError(ctx, op, err, event.WithInfoMsg("error getting current controllers"))
 		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error getting current controllers: %v", err)

internal/daemon/cluster/handlers/worker_service_status_test.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"sort"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -95,7 +96,7 @@ func TestStatus(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, err)
 
-	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
+	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64))
 	require.NotNil(t, s)
 
 	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId)
@@ -299,7 +300,7 @@ func TestStatusSessionClosed(t *testing.T) {
 	sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, tofu2)
 	require.NoError(t, err)
 
-	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
+	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64))
 	require.NotNil(t, s)
 
 	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId)
@@ -483,7 +484,7 @@ func TestStatusDeadConnection(t *testing.T) {
 	sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, tofu2)
 	require.NoError(t, err)
 
-	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
+	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64))
 	require.NotNil(t, s)
 
 	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId)
@@ -641,7 +642,7 @@ func TestStatusWorkerWithKeyId(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, err)
 
-	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
+	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64))
 	require.NotNil(t, s)
 
 	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId)
@@ -782,7 +783,7 @@ func TestStatusAuthorizedWorkers(t *testing.T) {
 	_ = server.TestPkiWorker(t, conn, wrapper, server.WithTestPkiWorkerAuthorizedKeyId(&w1KeyId))
 	_ = server.TestPkiWorker(t, conn, wrapper, server.WithTestPkiWorkerAuthorizedKeyId(&w2KeyId))
 
-	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kmsCache)
+	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kmsCache, new(atomic.Int64))
 	require.NotNil(t, s)
 
 	cases := []struct {
@@ -923,7 +924,7 @@ func TestWorkerOperationalStatus(t *testing.T) {
 
 	worker1 := server.TestKmsWorker(t, conn, wrapper)
 
-	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
+	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64))
 	require.NotNil(t, s)
 
 	cases := []struct {

internal/daemon/cluster/handlers/worker_service_test.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -125,7 +126,7 @@ func TestLookupSession(t *testing.T) {
 	err = repo.AddSessionCredentials(ctx, sessWithCreds.ProjectId, sessWithCreds.GetPublicId(), workerCreds)
 	require.NoError(t, err)
 
-	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kms)
+	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kms, new(atomic.Int64))
 	require.NotNil(t, s)
 
 	cases := []struct {
@@ -237,7 +238,7 @@ func TestHcpbWorkers(t *testing.T) {
 		server.TestPkiWorker(t, conn, wrapper, opt...)
 	}
 
-	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kmsCache)
+	s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kmsCache, new(atomic.Int64))
 	require.NotNil(t, s)
 
 	res, err := s.ListHcpbWorkers(ctx, &pbs.ListHcpbWorkersRequest{})

internal/daemon/controller/controller.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	"github.com/hashicorp/boundary/internal/auth/oidc"
 	"github.com/hashicorp/boundary/internal/auth/password"
@@ -104,6 +105,11 @@ type Controller struct {
 	// Used for testing and tracking worker health
 	workerStatusUpdateTimes *sync.Map
 
+	// Timing variables. These are atomics for SIGHUP support, and are int64
+	// because they are casted to time.Duration.
+	workerStatusGracePeriod *atomic.Int64
+	livenessTimeToStale     *atomic.Int64
+
 	apiGrpcServer         *grpc.Server
 	apiGrpcServerListener grpcServerListener
 	apiGrpcGatewayTicket  string
@@ -150,6 +156,8 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 		enabledPlugins:          conf.Server.EnabledPlugins,
 		apiListeners:            make([]*base.ServerListener, 0),
 		pkiConnManager:          cluster.NewDownstreamManager(),
+		workerStatusGracePeriod: new(atomic.Int64),
+		livenessTimeToStale:     new(atomic.Int64),
 	}
 
 	if downstreamRouterFactory != nil {
@@ -187,6 +195,19 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 		}
 	}
 
+	switch conf.RawConfig.Controller.WorkerStatusGracePeriodDuration {
+	case 0:
+		c.workerStatusGracePeriod.Store(int64(server.DefaultLiveness))
+	default:
+		c.workerStatusGracePeriod.Store(int64(conf.RawConfig.Controller.WorkerStatusGracePeriodDuration))
+	}
+	switch conf.RawConfig.Controller.LivenessTimeToStaleDuration {
+	case 0:
+		c.livenessTimeToStale.Store(int64(server.DefaultLiveness))
+	default:
+		c.livenessTimeToStale.Store(int64(conf.RawConfig.Controller.LivenessTimeToStaleDuration))
+	}
+
 	clusterListeners := make([]*base.ServerListener, 0)
 	for i := range conf.Listeners {
 		l := conf.Listeners[i]
@@ -487,7 +508,7 @@ func (c *Controller) registerJobs() error {
 	if err := pluginhost.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms, c.conf.HostPlugins); err != nil {
 		return err
 	}
-	if err := session.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms, c.conf.StatusGracePeriodDuration); err != nil {
+	if err := session.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms, c.workerStatusGracePeriod); err != nil {
 		return err
 	}
 	if err := serversjob.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms); err != nil {

internal/daemon/controller/handler.go

@@ -177,7 +177,8 @@ func (c *Controller) registerGrpcServices(s *grpc.Server) error {
 			c.PluginHostRepoFn,
 			c.StaticHostRepoFn,
 			c.VaultCredentialRepoFn,
-			c.StaticCredentialRepoFn)
+			c.StaticCredentialRepoFn,
+			c.workerStatusGracePeriod)
 		if err != nil {
 			return fmt.Errorf("failed to create target handler service: %w", err)
 		}

internal/daemon/controller/handlers/targets/target_service.go

@@ -9,6 +9,8 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync/atomic"
+	"time"
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/credential"
@@ -95,15 +97,16 @@ var (
 type Service struct {
 	pbs.UnsafeTargetServiceServer
 
-	repoFn           target.RepositoryFactory
-	iamRepoFn        common.IamRepoFactory
-	serversRepoFn    common.ServersRepoFactory
-	sessionRepoFn    session.RepositoryFactory
-	pluginHostRepoFn common.PluginHostRepoFactory
-	staticHostRepoFn common.StaticRepoFactory
-	vaultCredRepoFn  common.VaultCredentialRepoFactory
-	staticCredRepoFn common.StaticCredentialRepoFactory
-	kmsCache         *kms.Kms
+	repoFn                  target.RepositoryFactory
+	iamRepoFn               common.IamRepoFactory
+	serversRepoFn           common.ServersRepoFactory
+	sessionRepoFn           session.RepositoryFactory
+	pluginHostRepoFn        common.PluginHostRepoFactory
+	staticHostRepoFn        common.StaticRepoFactory
+	vaultCredRepoFn         common.VaultCredentialRepoFactory
+	staticCredRepoFn        common.StaticCredentialRepoFactory
+	kmsCache                *kms.Kms
+	workerStatusGracePeriod *atomic.Int64
 }
 
 var _ pbs.TargetServiceServer = (*Service)(nil)
@@ -120,6 +123,7 @@ func NewService(
 	staticHostRepoFn common.StaticRepoFactory,
 	vaultCredRepoFn common.VaultCredentialRepoFactory,
 	staticCredRepoFn common.StaticCredentialRepoFactory,
+	workerStatusGracePeriod *atomic.Int64,
 ) (Service, error) {
 	const op = "targets.NewService"
 	if repoFn == nil {
@@ -147,15 +151,16 @@ func NewService(
 		return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing static credential repository")
 	}
 	return Service{
-		repoFn:           repoFn,
-		iamRepoFn:        iamRepoFn,
-		serversRepoFn:    serversRepoFn,
-		sessionRepoFn:    sessionRepoFn,
-		pluginHostRepoFn: pluginHostRepoFn,
-		staticHostRepoFn: staticHostRepoFn,
-		vaultCredRepoFn:  vaultCredRepoFn,
-		staticCredRepoFn: staticCredRepoFn,
-		kmsCache:         kmsCache,
+		repoFn:                  repoFn,
+		iamRepoFn:               iamRepoFn,
+		serversRepoFn:           serversRepoFn,
+		sessionRepoFn:           sessionRepoFn,
+		pluginHostRepoFn:        pluginHostRepoFn,
+		staticHostRepoFn:        staticHostRepoFn,
+		vaultCredRepoFn:         vaultCredRepoFn,
+		staticCredRepoFn:        staticCredRepoFn,
+		kmsCache:                kmsCache,
+		workerStatusGracePeriod: workerStatusGracePeriod,
 	}, nil
 }
 
@@ -681,7 +686,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 	// worker IDs below is used to contain their IDs in the same order. This is
 	// used to fetch tags for filtering. But we avoid allocation unless we
 	// actually need it.
-	selectedWorkers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithActiveWorkers(true))
+	selectedWorkers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithLiveness(time.Duration(s.workerStatusGracePeriod.Load())))
 	if err != nil {
 		return nil, err
 	}

internal/daemon/controller/handlers/targets/tcp/target_service_test.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"path"
 	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -78,6 +79,8 @@ var testAuthorizedActions = []string{
 func testService(t *testing.T, ctx context.Context, conn *db.DB, kms *kms.Kms, wrapper wrapping.Wrapper) (targets.Service, error) {
 	rw := db.New(conn)
 	sche := scheduler.TestScheduler(t, conn, wrapper)
+	statusGracePeriod := new(atomic.Int64)
+	statusGracePeriod.Store(int64(server.DefaultLiveness))
 	repoFn := func(o ...target.Option) (*target.Repository, error) {
 		return target.NewRepository(ctx, rw, rw, kms, o...)
 	}
@@ -102,7 +105,7 @@ func testService(t *testing.T, ctx context.Context, conn *db.DB, kms *kms.Kms, w
 	staticCredRepoFn := func() (*credstatic.Repository, error) {
 		return credstatic.NewRepository(context.Background(), rw, rw, kms)
 	}
-	return targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn)
+	return targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod)
 }
 
 func TestGet(t *testing.T) {
@@ -2406,7 +2409,9 @@ func TestAuthorizeSession(t *testing.T) {
 		},
 	}
 
-	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn)
+	statusGracePeriod := new(atomic.Int64)
+	statusGracePeriod.Store(int64(server.DefaultLiveness))
+	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod)
 	require.NoError(t, err)
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -2571,7 +2576,9 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) {
 	_ = iam.TestUserRole(t, conn, r.GetPublicId(), at.GetIamUserId())
 	_ = iam.TestRoleGrant(t, conn, r.GetPublicId(), "id=*;type=*;actions=*")
 
-	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn)
+	statusGracePeriod := new(atomic.Int64)
+	statusGracePeriod.Store(int64(server.DefaultLiveness))
+	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod)
 	require.NoError(t, err)
 
 	hc := static.TestCatalogs(t, conn, proj.GetPublicId(), 1)[0]
@@ -3134,7 +3141,9 @@ func TestAuthorizeSession_Errors(t *testing.T) {
 	}
 	org, proj := iam.TestScopes(t, iamRepo)
 
-	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn)
+	statusGracePeriod := new(atomic.Int64)
+	statusGracePeriod.Store(int64(server.DefaultLiveness))
+	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod)
 	require.NoError(t, err)
 
 	// Authorized user gets full permissions

internal/daemon/controller/rpc_registration.go

@@ -34,7 +34,7 @@ func registerControllerServerCoordinationService(ctx context.Context, c *Control
 	}
 
 	workerService := handlers.NewWorkerServiceServer(c.ServersRepoFn, c.WorkerAuthRepoStorageFn,
-		c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms)
+		c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms, c.livenessTimeToStale)
 	pbs.RegisterServerCoordinationServiceServer(server, workerService)
 	return nil
 }
@@ -52,7 +52,7 @@ func registerControllerSessionService(ctx context.Context, c *Controller, server
 	}
 
 	workerService := handlers.NewWorkerServiceServer(c.ServersRepoFn, c.WorkerAuthRepoStorageFn,
-		c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms)
+		c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms, c.livenessTimeToStale)
 	pbs.RegisterSessionServiceServer(server, workerService)
 	return nil
 }

internal/daemon/controller/testing.go

@@ -423,9 +423,14 @@ type TestControllerOpts struct {
 	// (overrides address provided in config, if any)
 	PublicClusterAddr string
 
-	// The amount of time to wait before marking connections as closed when a
+	// The amount of time to wait before marking connections as canceling when a
 	// worker has not reported in
-	StatusGracePeriodDuration time.Duration
+	WorkerStatusGracePeriodDuration time.Duration
+
+	// The period of time after which it will consider other controllers to be
+	// no longer accessible, based on time since their last status update in the
+	// database
+	LivenessTimeToStaleDuration time.Duration
 
 	// The amount of time between the scheduler waking up to run it's registered
 	// jobs.
@@ -612,8 +617,12 @@ func TestControllerConfig(t testing.TB, ctx context.Context, tc *TestController,
 		}
 	}
 
-	// Initialize status grace period
-	tc.b.SetStatusGracePeriodDuration(opts.StatusGracePeriodDuration)
+	if opts.WorkerStatusGracePeriodDuration != 0 {
+		opts.Config.Controller.WorkerStatusGracePeriodDuration = opts.WorkerStatusGracePeriodDuration
+	}
+	if opts.LivenessTimeToStaleDuration != 0 {
+		opts.Config.Controller.LivenessTimeToStaleDuration = opts.LivenessTimeToStaleDuration
+	}
 
 	tc.name = opts.Config.Controller.Name
 
@@ -738,20 +747,21 @@ func (tc *TestController) AddClusterControllerMember(t testing.TB, opts *TestCon
 		opts = new(TestControllerOpts)
 	}
 	nextOpts := &TestControllerOpts{
-		DatabaseUrl:                 tc.c.conf.DatabaseUrl,
-		DefaultPasswordAuthMethodId: tc.c.conf.DevPasswordAuthMethodId,
-		DefaultOidcAuthMethodId:     tc.c.conf.DevOidcAuthMethodId,
-		RootKms:                     tc.c.conf.RootKms,
-		WorkerAuthKms:               tc.c.conf.WorkerAuthKms,
-		RecoveryKms:                 tc.c.conf.RecoveryKms,
-		Name:                        opts.Name,
-		Logger:                      tc.c.conf.Logger,
-		DefaultLoginName:            tc.b.DevLoginName,
-		DefaultPassword:             tc.b.DevPassword,
-		DisableKmsKeyCreation:       true,
-		DisableAuthMethodCreation:   true,
-		PublicClusterAddr:           opts.PublicClusterAddr,
-		StatusGracePeriodDuration:   opts.StatusGracePeriodDuration,
+		DatabaseUrl:                     tc.c.conf.DatabaseUrl,
+		DefaultPasswordAuthMethodId:     tc.c.conf.DevPasswordAuthMethodId,
+		DefaultOidcAuthMethodId:         tc.c.conf.DevOidcAuthMethodId,
+		RootKms:                         tc.c.conf.RootKms,
+		WorkerAuthKms:                   tc.c.conf.WorkerAuthKms,
+		RecoveryKms:                     tc.c.conf.RecoveryKms,
+		Name:                            opts.Name,
+		Logger:                          tc.c.conf.Logger,
+		DefaultLoginName:                tc.b.DevLoginName,
+		DefaultPassword:                 tc.b.DevPassword,
+		DisableKmsKeyCreation:           true,
+		DisableAuthMethodCreation:       true,
+		PublicClusterAddr:               opts.PublicClusterAddr,
+		WorkerStatusGracePeriodDuration: opts.WorkerStatusGracePeriodDuration,
+		LivenessTimeToStaleDuration:     opts.LivenessTimeToStaleDuration,
 	}
 	if opts.Logger != nil {
 		nextOpts.Logger = opts.Logger
@@ -775,7 +785,7 @@ func (tc *TestController) WaitForNextWorkerStatusUpdate(workerStatusName string)
 	ctx := context.TODO()
 	event.WriteSysEvent(ctx, op, "waiting for next status report from worker", "worker", workerStatusName)
 	waitStatusStart := time.Now()
-	ctx, cancel := context.WithTimeout(tc.ctx, tc.b.StatusGracePeriodDuration)
+	ctx, cancel := context.WithTimeout(tc.ctx, time.Duration(tc.c.workerStatusGracePeriod.Load()))
 	defer cancel()
 	var err error
 	for {

internal/daemon/worker/common/common.go

@@ -1,6 +1,10 @@
 package common
 
-import "time"
+import (
+	"time"
+
+	"github.com/hashicorp/boundary/internal/server"
+)
 
 // In the future we could make this configurable
 const (
@@ -8,7 +12,7 @@ const (
 	// during the worker status report
 	StatusInterval = 2 * time.Second
 
-	// StatusTimeout is the timeout duration on status calls to the controller from
+	// DefaultStatusTimeout is the timeout duration on status calls to the controller from
 	// the worker
-	StatusTimeout = 5 * time.Second
+	DefaultStatusTimeout = server.DefaultLiveness / 3
 )

internal/daemon/worker/session/manager.go

@@ -5,11 +5,19 @@ import (
 	"fmt"
 	"reflect"
 	"sync"
+	"sync/atomic"
 
 	"github.com/hashicorp/boundary/internal/errors"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 )
 
+// FIXME: This is really ugly -- but not as ugly as plumbing this value into the
+// interface. We should figure out something better. For now, it will at least
+// keep sync with any changes to the value once it's initialized as the worker
+// will set it in its new function; it can also be overridden in tests if
+// desired with normal global variable caveats.
+var CloseCallTimeout = new(atomic.Int64)
+
 // Manager stores session information locally and exposes ways to operate on the
 // set of sessions locally in batch.
 // This is thread-safe.

internal/daemon/worker/session/session.go

@@ -9,7 +9,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/hashicorp/boundary/internal/daemon/worker/common"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/session"
@@ -503,13 +502,14 @@ func closeConnections(ctx context.Context, sessClient pbs.SessionServiceClient,
 	// bit of formalization in terms of how we handle timeouts. For now, this
 	// just ensures consistency with the same status call in that it times out
 	// within an adequate period of time.
-	closeConnCtx, closeConnCancel := context.WithTimeout(ctx, common.StatusTimeout)
+	closeConnCtx, closeConnCancel := context.WithTimeout(ctx, time.Duration(CloseCallTimeout.Load()))
 	defer closeConnCancel()
 	response, err := closeConnection(closeConnCtx, sessClient, makeCloseConnectionRequest(closeInfo))
 	if err != nil {
 		event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connections closed",
 			"warning", "error contacting controller, connections will be closed only on worker",
 			"session_and_connection_ids", fmt.Sprintf("%#v", closeInfo),
+			"timeout", time.Duration(CloseCallTimeout.Load()).String(),
 		))
 
 		// Since we could not reach the controller, we have to make a "fake" response set.

internal/daemon/worker/status.go

@@ -86,7 +86,7 @@ func (w *Worker) LastStatusSuccess() *LastStatusInformation {
 func (w *Worker) WaitForNextSuccessfulStatusUpdate() error {
 	const op = "worker.(Worker).WaitForNextSuccessfulStatusUpdate"
 	waitStatusStart := time.Now()
-	ctx, cancel := context.WithTimeout(w.baseContext, w.conf.StatusGracePeriodDuration)
+	ctx, cancel := context.WithTimeout(w.baseContext, time.Duration(w.successfulStatusGracePeriod.Load()))
 	defer cancel()
 	event.WriteSysEvent(ctx, op, "waiting for next status report to controller")
 	for {
@@ -122,8 +122,9 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 		var jobInfo pbs.SessionJobInfo
 		status := s.GetStatus()
 		sessionId := s.GetId()
-		connections := make([]*pbs.Connection, 0, len(s.GetLocalConnections()))
-		for k, v := range s.GetLocalConnections() {
+		localConnections := s.GetLocalConnections()
+		connections := make([]*pbs.Connection, 0, len(localConnections))
+		for k, v := range localConnections {
 			connections = append(connections, &pbs.Connection{
 				ConnectionId: k,
 				Status:       v.Status,
@@ -155,7 +156,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 	if w.updateTags.Load() {
 		tags = w.tags.Load().([]*pb.TagPair)
 	}
-	statusCtx, statusCancel := context.WithTimeout(cancelCtx, common.StatusTimeout)
+	statusCtx, statusCancel := context.WithTimeout(cancelCtx, time.Duration(w.statusCallTimeoutDuration.Load()))
 	defer statusCancel()
 
 	keyId := w.WorkerAuthCurrentKeyId.Load()
@@ -229,7 +230,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 		}
 	} else if w.conf.RawConfig.HcpbClusterId != "" && len(w.conf.RawConfig.Worker.InitialUpstreams) == 0 {
 		// This is a worker that is one hop away from managed workers, so attempt to get that list
-		hcpbWorkersCtx, hcpbWorkersCancel := context.WithTimeout(cancelCtx, common.StatusTimeout)
+		hcpbWorkersCtx, hcpbWorkersCancel := context.WithTimeout(cancelCtx, time.Duration(w.statusCallTimeoutDuration.Load()))
 		defer hcpbWorkersCancel()
 		workersResp, err := client.ListHcpbWorkers(hcpbWorkersCtx, &pbs.ListHcpbWorkersRequest{})
 		if err != nil {
@@ -409,7 +410,7 @@ func (w *Worker) lastSuccessfulStatusTime() time.Time {
 
 func (w *Worker) isPastGrace() (bool, time.Time, time.Duration) {
 	t := w.lastSuccessfulStatusTime()
-	u := w.conf.StatusGracePeriodDuration
+	u := time.Duration(w.successfulStatusGracePeriod.Load())
 	v := time.Since(t)
 	return v > u, t, u
 }

internal/daemon/worker/status_test.go

@@ -27,18 +27,18 @@ func TestWorkerWaitForNextSuccessfulStatusUpdate(t *testing.T) {
 
 			// As-needed initialization of a mock worker
 			w := &Worker{
-				logger:            hclog.New(nil),
-				lastStatusSuccess: new(atomic.Value),
-				baseContext:       context.Background(),
+				logger:                      hclog.New(nil),
+				lastStatusSuccess:           new(atomic.Value),
+				baseContext:                 context.Background(),
+				successfulStatusGracePeriod: new(atomic.Int64),
 				conf: &Config{
-					Server: &base.Server{
-						StatusGracePeriodDuration: time.Second * 2,
-					},
+					Server: &base.Server{},
 				},
 			}
 
 			// This is present in New()
 			w.lastStatusSuccess.Store((*LastStatusInformation)(nil))
+			w.successfulStatusGracePeriod.Store(int64(time.Second * 2))
 
 			var wg sync.WaitGroup
 			var err error

internal/daemon/worker/testing.go

@@ -204,7 +204,7 @@ type TestWorkerOpts struct {
 
 	// The amount of time to wait before marking connections as closed when a
 	// connection cannot be made back to the controller
-	StatusGracePeriodDuration time.Duration
+	SuccessfulStatusGracePeriodDuration time.Duration
 
 	// Overrides worker's nonceFn, for cases where we want to have control
 	// over the nonce we send to the Controller
@@ -263,9 +263,6 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 
 	tw.b.PrometheusRegisterer = opts.PrometheusRegisterer
 
-	// Initialize status grace period
-	tw.b.SetStatusGracePeriodDuration(opts.StatusGracePeriodDuration)
-
 	if opts.Config.Worker == nil {
 		opts.Config.Worker = &config.Worker{
 			Name: opts.Name,
@@ -277,6 +274,10 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 	}
 	tw.name = opts.Config.Worker.Name
 
+	if opts.SuccessfulStatusGracePeriodDuration != 0 {
+		opts.Config.Worker.SuccessfulStatusGracePeriod = opts.SuccessfulStatusGracePeriodDuration
+	}
+
 	serverName, err := os.Hostname()
 	if err != nil {
 		t.Fatal(err)
@@ -357,12 +358,12 @@ func (tw *TestWorker) AddClusterWorkerMember(t testing.TB, opts *TestWorkerOpts)
 		opts = new(TestWorkerOpts)
 	}
 	nextOpts := &TestWorkerOpts{
-		WorkerAuthKms:             tw.w.conf.WorkerAuthKms,
-		WorkerAuthStorageKms:      tw.w.conf.WorkerAuthStorageKms,
-		Name:                      opts.Name,
-		InitialUpstreams:          tw.UpstreamAddrs(),
-		Logger:                    tw.w.conf.Logger,
-		StatusGracePeriodDuration: opts.StatusGracePeriodDuration,
+		WorkerAuthKms:                       tw.w.conf.WorkerAuthKms,
+		WorkerAuthStorageKms:                tw.w.conf.WorkerAuthStorageKms,
+		Name:                                opts.Name,
+		InitialUpstreams:                    tw.UpstreamAddrs(),
+		Logger:                              tw.w.conf.Logger,
+		SuccessfulStatusGracePeriodDuration: opts.SuccessfulStatusGracePeriodDuration,
 	}
 	if nextOpts.Name == "" {
 		var err error

internal/daemon/worker/worker.go

@@ -17,6 +17,7 @@ import (
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	"github.com/hashicorp/boundary/internal/daemon/cluster"
+	"github.com/hashicorp/boundary/internal/daemon/worker/common"
 	"github.com/hashicorp/boundary/internal/daemon/worker/internal/metric"
 	"github.com/hashicorp/boundary/internal/daemon/worker/session"
 	"github.com/hashicorp/boundary/internal/errors"
@@ -128,6 +129,11 @@ type Worker struct {
 	downstreamWorkers downstreamers
 	downstreamRoutes  downstreamRouter
 
+	// Timing variables. These are atomics for SIGHUP support, and are int64
+	// because they are casted to time.Duration.
+	successfulStatusGracePeriod *atomic.Int64
+	statusCallTimeoutDuration   *atomic.Int64
+
 	// Test-specific options (and possibly hidden dev-mode flags)
 	TestOverrideX509VerifyDnsName  string
 	TestOverrideX509VerifyCertPool *x509.CertPool
@@ -146,19 +152,21 @@ func New(conf *Config) (*Worker, error) {
 	initializeReverseGrpcClientCollectors(conf.PrometheusRegisterer)
 
 	w := &Worker{
-		conf:                   conf,
-		logger:                 conf.Logger.Named("worker"),
-		started:                ua.NewBool(false),
-		controllerStatusConn:   new(atomic.Value),
-		everAuthenticated:      ua.NewUint32(authenticationStatusNeverAuthenticated),
-		lastStatusSuccess:      new(atomic.Value),
-		controllerMultihopConn: new(atomic.Value),
-		tags:                   new(atomic.Value),
-		updateTags:             ua.NewBool(false),
-		nonceFn:                base62.Random,
-		WorkerAuthCurrentKeyId: new(ua.String),
-		operationalState:       new(atomic.Value),
-		pkiConnManager:         cluster.NewDownstreamManager(),
+		conf:                        conf,
+		logger:                      conf.Logger.Named("worker"),
+		started:                     ua.NewBool(false),
+		controllerStatusConn:        new(atomic.Value),
+		everAuthenticated:           ua.NewUint32(authenticationStatusNeverAuthenticated),
+		lastStatusSuccess:           new(atomic.Value),
+		controllerMultihopConn:      new(atomic.Value),
+		tags:                        new(atomic.Value),
+		updateTags:                  ua.NewBool(false),
+		nonceFn:                     base62.Random,
+		WorkerAuthCurrentKeyId:      new(ua.String),
+		operationalState:            new(atomic.Value),
+		pkiConnManager:              cluster.NewDownstreamManager(),
+		successfulStatusGracePeriod: new(atomic.Int64),
+		statusCallTimeoutDuration:   new(atomic.Int64),
 	}
 
 	if downstreamRouterFactory != nil {
@@ -196,6 +204,21 @@ func New(conf *Config) (*Worker, error) {
 		}
 	}
 
+	switch conf.RawConfig.Worker.SuccessfulStatusGracePeriodDuration {
+	case 0:
+		w.successfulStatusGracePeriod.Store(int64(server.DefaultLiveness))
+	default:
+		w.successfulStatusGracePeriod.Store(int64(conf.RawConfig.Worker.SuccessfulStatusGracePeriodDuration))
+	}
+	switch conf.RawConfig.Worker.StatusCallTimeoutDuration {
+	case 0:
+		w.statusCallTimeoutDuration.Store(int64(common.DefaultStatusTimeout))
+	default:
+		w.statusCallTimeoutDuration.Store(int64(conf.RawConfig.Worker.StatusCallTimeoutDuration))
+	}
+	// FIXME: This is really ugly, but works.
+	session.CloseCallTimeout = w.statusCallTimeoutDuration
+
 	var listenerCount int
 	for i := range conf.Listeners {
 		l := conf.Listeners[i]
@@ -502,11 +525,11 @@ func (w *Worker) Shutdown() error {
 	// Shut down all connections.
 	w.cleanupConnections(w.baseContext, true, w.sessionManager)
 
-	// Wait for next status request to succeed. Don't wait too long;
-	// wrap the base context in a timeout equal to our status grace
-	// period.
+	// Wait for next status request to succeed. Don't wait too long; time it out
+	// at our default liveness value, which is also our default status grace
+	// period timeout
 	waitStatusStart := time.Now()
-	nextStatusCtx, nextStatusCancel := context.WithTimeout(w.baseContext, w.conf.StatusGracePeriodDuration)
+	nextStatusCtx, nextStatusCancel := context.WithTimeout(w.baseContext, server.DefaultLiveness)
 	defer nextStatusCancel()
 	for {
 		if err := nextStatusCtx.Err(); err != nil {

internal/server/repository.go

@@ -10,14 +10,8 @@ import (
 )
 
 const (
-	// DefaultLiveness is the setting that controls the server "liveness" time,
-	// or the maximum allowable time that a worker can't send a status update to
-	// the controller for. After this, the server is considered dead, and it will
-	// be taken out of the rotation for allowable workers for connections, and
-	// connections will possibly start to be terminated and marked as closed
-	// depending on the grace period setting (see
-	// base.Server.StatusGracePeriodDuration). This value serves as the default
-	// and minimum allowable setting for the grace period.
+	// DefaultLiveness is a default used for various timing parameters, such as
+	// grace period for status updates, server liveness, etc.
 	DefaultLiveness = 15 * time.Second
 )
 

internal/session/job_session_cleanup.go

@@ -4,21 +4,15 @@ import (
 	"context"
 	"database/sql"
 	stderrors "errors"
-	"fmt"
+	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/scheduler"
-	"github.com/hashicorp/boundary/internal/server"
 )
 
-// deadWorkerConnCloseMinGrace is the minimum allowable setting for
-// the closeConnectionsForDeadWorkers method. This is synced with
-// the default server liveness setting.
-const deadWorkerConnCloseMinGrace = server.DefaultLiveness
-
 type closeConnectionsForDeadWorkersResult struct {
 	WorkerId                string
 	LastUpdateTime          time.Time
@@ -37,10 +31,10 @@ type closeConnectionsForDeadWorkersResult struct {
 type sessionConnectionCleanupJob struct {
 	writer db.Writer
 
-	// The amount of time to give disconnected workers before marking
-	// their connections as closed.  This should be at larger than the liveness
-	// setting for the worker.
-	gracePeriod time.Duration
+	// The amount of time to give disconnected workers before marking their
+	// connections as closed. This should be larger than the liveness setting
+	// for the worker.
+	gracePeriod *atomic.Int64
 
 	// The total number of connections closed in the last run.
 	totalClosed int
@@ -49,15 +43,16 @@ type sessionConnectionCleanupJob struct {
 // newSessionConnectionCleanupJob instantiates the session cleanup job.
 func newSessionConnectionCleanupJob(
 	writer db.Writer,
-	gracePeriod time.Duration,
+	gracePeriod *atomic.Int64,
 ) (*sessionConnectionCleanupJob, error) {
 	const op = "session.newNewSessionConnectionCleanupJob"
 	switch {
 	case writer == nil:
 		return nil, errors.NewDeprecated(errors.InvalidParameter, op, "missing db writer")
-	case gracePeriod < deadWorkerConnCloseMinGrace:
-		return nil, errors.NewDeprecated(
-			errors.InvalidParameter, op, fmt.Sprintf("invalid gracePeriod, must be greater than %s", deadWorkerConnCloseMinGrace))
+	case gracePeriod == nil:
+		return nil, errors.NewDeprecated(errors.InvalidParameter, op, "missing grace period")
+	case gracePeriod.Load() == 0:
+		return nil, errors.NewDeprecated(errors.InvalidParameter, op, "grace period is zero")
 	}
 
 	return &sessionConnectionCleanupJob{
@@ -99,7 +94,8 @@ func (j *sessionConnectionCleanupJob) Run(ctx context.Context) error {
 	j.totalClosed = 0
 
 	// Run the atomic dead worker cleanup job.
-	results, err := j.closeConnectionsForDeadWorkers(ctx, j.gracePeriod)
+	gracePeriod := time.Duration(j.gracePeriod.Load())
+	results, err := j.closeConnectionsForDeadWorkers(ctx, gracePeriod)
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
@@ -109,7 +105,7 @@ func (j *sessionConnectionCleanupJob) Run(ctx context.Context) error {
 			event.WithInfo(
 				"public_id", result.WorkerId,
 				"update_time", result.LastUpdateTime,
-				"grace_period_seconds", j.gracePeriod.Seconds(),
+				"grace_period_seconds", gracePeriod.Seconds(),
 				"number_connections_closed", result.NumberConnectionsClosed,
 			))
 

internal/session/job_session_cleanup_test.go

@@ -3,6 +3,7 @@ package session
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -27,7 +28,8 @@ func TestSessionConnectionCleanupJob(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	const gracePeriod = 1 * time.Second
+	gracePeriod := new(atomic.Int64)
+	gracePeriod.Store(int64(time.Second))
 
 	require, assert := require.New(t), assert.New(t)
 	conn, _ := db.TestSetup(t, "postgres")
@@ -100,12 +102,12 @@ func TestSessionConnectionCleanupJob(t *testing.T) {
 	}
 
 	// Create the job.
-	job, err := newSessionConnectionCleanupJob(rw, deadWorkerConnCloseMinGrace)
+	job, err := newSessionConnectionCleanupJob(rw, gracePeriod)
 	job.gracePeriod = gracePeriod // by-pass factory assert so we dont have to wait so long
 	require.NoError(err)
 
 	// sleep the status grace period.
-	time.Sleep(gracePeriod)
+	time.Sleep(time.Duration(gracePeriod.Load()))
 
 	// Push an upsert to the first worker so that its status has been
 	// updated.
@@ -149,7 +151,10 @@ func TestSessionConnectionCleanupJobNewJobErr(t *testing.T) {
 	const op = "session.newNewSessionConnectionCleanupJob"
 	require := require.New(t)
 
-	job, err := newSessionConnectionCleanupJob(nil, 0)
+	grace := new(atomic.Int64)
+	grace.Store(1000000)
+
+	job, err := newSessionConnectionCleanupJob(nil, grace)
 	require.Equal(err, errors.E(
 		ctx,
 		errors.WithCode(errors.InvalidParameter),
@@ -161,12 +166,21 @@ func TestSessionConnectionCleanupJobNewJobErr(t *testing.T) {
 	conn, _ := db.TestSetup(t, "postgres")
 	rw := db.New(conn)
 
-	job, err = newSessionConnectionCleanupJob(rw, 0)
+	job, err = newSessionConnectionCleanupJob(rw, nil)
 	require.Equal(err, errors.E(
 		ctx,
 		errors.WithCode(errors.InvalidParameter),
 		errors.WithOp(op),
-		errors.WithMsg(fmt.Sprintf("invalid gracePeriod, must be greater than %s", deadWorkerConnCloseMinGrace)),
+		errors.WithMsg(fmt.Sprintf("missing grace period")),
+	))
+	require.Nil(job)
+
+	job, err = newSessionConnectionCleanupJob(rw, new(atomic.Int64))
+	require.Equal(err, errors.E(
+		ctx,
+		errors.WithCode(errors.InvalidParameter),
+		errors.WithOp(op),
+		errors.WithMsg(fmt.Sprintf("grace period is zero")),
 	))
 	require.Nil(job)
 }
@@ -188,7 +202,10 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) {
 	serversRepo, err := server.NewRepository(rw, rw, kms)
 	require.NoError(err)
 
-	job, err := newSessionConnectionCleanupJob(rw, deadWorkerConnCloseMinGrace)
+	defaultLiveness := new(atomic.Int64)
+	defaultLiveness.Store(int64(server.DefaultLiveness))
+
+	job, err := newSessionConnectionCleanupJob(rw, defaultLiveness)
 	require.NoError(err)
 
 	// connection count = 6 * states(authorized, connected, closed = 3) * servers_with_open_connections(3)
@@ -448,7 +465,10 @@ func TestCloseWorkerlessConnections(t *testing.T) {
 	connRepo, err := NewConnectionRepository(ctx, rw, rw, kms)
 	require.NoError(err)
 
-	job, err := newSessionConnectionCleanupJob(rw, time.Hour)
+	hourDuration := new(atomic.Int64)
+	hourDuration.Store(int64(time.Hour))
+
+	job, err := newSessionConnectionCleanupJob(rw, hourDuration)
 	require.NoError(err)
 
 	createConnection := func(workerId string) *Connection {

internal/session/jobs.go

@@ -3,9 +3,11 @@ package session
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/scheduler"
 )
@@ -13,9 +15,13 @@ import (
 const deleteTerminatedThreshold = time.Hour
 
 // RegisterJobs registers session related jobs with the provided scheduler.
-func RegisterJobs(ctx context.Context, scheduler *scheduler.Scheduler, w db.Writer, r db.Reader, k *kms.Kms, gracePeriod time.Duration) error {
+func RegisterJobs(ctx context.Context, scheduler *scheduler.Scheduler, w db.Writer, r db.Reader, k *kms.Kms, gracePeriod *atomic.Int64) error {
 	const op = "session.RegisterJobs"
 
+	if gracePeriod == nil {
+		return errors.New(ctx, errors.InvalidParameter, op, "nil grace period")
+	}
+
 	sessionConnectionCleanupJob, err := newSessionConnectionCleanupJob(w, gracePeriod)
 	if err != nil {
 		return fmt.Errorf("error creating session cleanup job: %w", err)

internal/tests/cluster/session_cleanup_test.go

@@ -46,14 +46,14 @@ var timeoutBurdenCases = []timeoutBurdenType{timeoutBurdenTypeDefault, timeoutBu
 
 func controllerGracePeriod(ty timeoutBurdenType) time.Duration {
 	if ty == timeoutBurdenTypeWorker {
-		return helper.DefaultGracePeriod * 10
+		return helper.DefaultWorkerStatusGracePeriod * 10
 	}
 
-	return helper.DefaultGracePeriod
+	return helper.DefaultWorkerStatusGracePeriod
 }
 
 func workerGracePeriod(ty timeoutBurdenType) time.Duration {
-	return helper.DefaultGracePeriod
+	return helper.DefaultWorkerStatusGracePeriod
 }
 
 // TestSessionCleanup is the main test for session cleanup, and
@@ -86,11 +86,11 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin
 		pl, err := net.Listen("tcp", "localhost:0")
 		require.NoError(err)
 		c1 := controller.NewTestController(t, &controller.TestControllerOpts{
-			Config:                    conf,
-			InitialResourcesSuffix:    "1234567890",
-			Logger:                    logger.Named("c1"),
-			PublicClusterAddr:         pl.Addr().String(),
-			StatusGracePeriodDuration: controllerGracePeriod(burdenCase),
+			Config:                          conf,
+			InitialResourcesSuffix:          "1234567890",
+			Logger:                          logger.Named("c1"),
+			PublicClusterAddr:               pl.Addr().String(),
+			WorkerStatusGracePeriodDuration: controllerGracePeriod(burdenCase),
 		})
 		defer c1.Shutdown()
 
@@ -108,10 +108,10 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin
 		require.NotEmpty(t, proxy.ListenerAddr())
 
 		w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{
-			WorkerAuthKms:             c1.Config().WorkerAuthKms,
-			InitialUpstreams:          []string{proxy.ListenerAddr()},
-			Logger:                    logger.Named("w1"),
-			StatusGracePeriodDuration: workerGracePeriod(burdenCase),
+			WorkerAuthKms:                       c1.Config().WorkerAuthKms,
+			InitialUpstreams:                    []string{proxy.ListenerAddr()},
+			Logger:                              logger.Named("w1"),
+			SuccessfulStatusGracePeriodDuration: workerGracePeriod(burdenCase),
 		})
 		defer w1.Shutdown()
 
@@ -216,11 +216,11 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing
 		pl1, err := net.Listen("tcp", "localhost:0")
 		require.NoError(err)
 		c1 := controller.NewTestController(t, &controller.TestControllerOpts{
-			Config:                    conf1,
-			InitialResourcesSuffix:    "1234567890",
-			Logger:                    logger.Named("c1"),
-			PublicClusterAddr:         pl1.Addr().String(),
-			StatusGracePeriodDuration: controllerGracePeriod(burdenCase),
+			Config:                          conf1,
+			InitialResourcesSuffix:          "1234567890",
+			Logger:                          logger.Named("c1"),
+			PublicClusterAddr:               pl1.Addr().String(),
+			WorkerStatusGracePeriodDuration: controllerGracePeriod(burdenCase),
 		})
 		defer c1.Shutdown()
 
@@ -230,9 +230,9 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing
 		pl2, err := net.Listen("tcp", "localhost:0")
 		require.NoError(err)
 		c2 := c1.AddClusterControllerMember(t, &controller.TestControllerOpts{
-			Logger:                    logger.Named("c2"),
-			PublicClusterAddr:         pl2.Addr().String(),
-			StatusGracePeriodDuration: controllerGracePeriod(burdenCase),
+			Logger:                          logger.Named("c2"),
+			PublicClusterAddr:               pl2.Addr().String(),
+			WorkerStatusGracePeriodDuration: controllerGracePeriod(burdenCase),
 		})
 		defer c2.Shutdown()
 		expectWorkers(t, c1)
@@ -268,10 +268,10 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing
 		// ** Worker **
 		// ************
 		w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{
-			WorkerAuthKms:             c1.Config().WorkerAuthKms,
-			InitialUpstreams:          []string{p1.ListenerAddr(), p2.ListenerAddr()},
-			Logger:                    logger.Named("w1"),
-			StatusGracePeriodDuration: workerGracePeriod(burdenCase),
+			WorkerAuthKms:                       c1.Config().WorkerAuthKms,
+			InitialUpstreams:                    []string{p1.ListenerAddr(), p2.ListenerAddr()},
+			Logger:                              logger.Named("w1"),
+			SuccessfulStatusGracePeriodDuration: workerGracePeriod(burdenCase),
 		})
 		defer w1.Shutdown()
 

internal/tests/cluster/worker_bytesupdown_test.go

@@ -31,11 +31,11 @@ func TestWorkerBytesUpDown(t *testing.T) {
 	pl, err := net.Listen("tcp", "localhost:0")
 	require.NoError(err)
 	c1 := controller.NewTestController(t, &controller.TestControllerOpts{
-		Config:                    conf,
-		InitialResourcesSuffix:    "1234567890",
-		Logger:                    logger.Named("c1"),
-		PublicClusterAddr:         pl.Addr().String(),
-		StatusGracePeriodDuration: helper.DefaultGracePeriod,
+		Config:                          conf,
+		InitialResourcesSuffix:          "1234567890",
+		Logger:                          logger.Named("c1"),
+		PublicClusterAddr:               pl.Addr().String(),
+		WorkerStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod,
 	})
 	t.Cleanup(c1.Shutdown)
 
@@ -53,10 +53,10 @@ func TestWorkerBytesUpDown(t *testing.T) {
 	t.Cleanup(func() { _ = proxy.Close() })
 
 	w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{
-		WorkerAuthKms:             c1.Config().WorkerAuthKms,
-		InitialUpstreams:          []string{proxy.ListenerAddr()},
-		Logger:                    logger.Named("w1"),
-		StatusGracePeriodDuration: helper.DefaultGracePeriod,
+		WorkerAuthKms:                       c1.Config().WorkerAuthKms,
+		InitialUpstreams:                    []string{proxy.ListenerAddr()},
+		Logger:                              logger.Named("w1"),
+		SuccessfulStatusGracePeriodDuration: helper.DefaultSuccessfulStatusGracePeriod,
 	})
 	t.Cleanup(w1.Shutdown)
 

internal/tests/cluster/worker_proxy_test.go

@@ -31,11 +31,11 @@ func TestWorkerSessionProxyMultipleConnections(t *testing.T) {
 	pl, err := net.Listen("tcp", "localhost:0")
 	require.NoError(err)
 	c1 := controller.NewTestController(t, &controller.TestControllerOpts{
-		Config:                    conf,
-		InitialResourcesSuffix:    "1234567890",
-		Logger:                    logger.Named("c1"),
-		PublicClusterAddr:         pl.Addr().String(),
-		StatusGracePeriodDuration: helper.DefaultGracePeriod,
+		Config:                          conf,
+		InitialResourcesSuffix:          "1234567890",
+		Logger:                          logger.Named("c1"),
+		PublicClusterAddr:               pl.Addr().String(),
+		WorkerStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod,
 	})
 	defer c1.Shutdown()
 
@@ -53,10 +53,10 @@ func TestWorkerSessionProxyMultipleConnections(t *testing.T) {
 	require.NotEmpty(t, proxy.ListenerAddr())
 
 	w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{
-		WorkerAuthKms:             c1.Config().WorkerAuthKms,
-		InitialUpstreams:          []string{proxy.ListenerAddr()},
-		Logger:                    logger.Named("w1"),
-		StatusGracePeriodDuration: helper.DefaultGracePeriod,
+		WorkerAuthKms:                       c1.Config().WorkerAuthKms,
+		InitialUpstreams:                    []string{proxy.ListenerAddr()},
+		Logger:                              logger.Named("w1"),
+		SuccessfulStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod,
 	})
 	defer w1.Shutdown()
 

internal/tests/helper/testing_helper.go

@@ -33,9 +33,10 @@ import (
 )
 
 const (
-	DefaultGracePeriod                       = time.Second * 15
+	DefaultWorkerStatusGracePeriod           = time.Second * 15
+	DefaultSuccessfulStatusGracePeriod       = time.Second * 15
 	expectConnectionStateOnControllerTimeout = time.Minute * 2
-	expectConnectionStateOnWorkerTimeout     = DefaultGracePeriod * 3
+	expectConnectionStateOnWorkerTimeout     = DefaultWorkerStatusGracePeriod * 3
 
 	// This is the interval that we check states on in the worker. It
 	// needs to be particularly granular to ensure that we allow for