Worker Domain Refactor (#5338)

* feat(server): Refactor LookupWorkerByName to be a test method

* feat(server): Split worker tags table

* refact(worker): Export api and config tags

* refact(server): Return worker id from UpsertWorkerStatus

* refact(server): Create ListWorkerStorageBucketUpdates domain function

* refact(server): Use dedicated HCPb-worker list func

* refact(server): Create ConnectionRoute domain function

* refact(handlers): Move worker selection logic to internal/server package

This commit moves all the logic to select a worker to handle a session
to the internal/server package. Most functions were moved to this
package with no changes, however, there are a couple of notable
exceptions:

- Given that we don't use GRPC errors in repos, all errors were adjusted
  to use the internal/errors package instead.
- StorageBucketFilterCredIdFn is now used to prevent the internal/server
  package from importing internal/storage/plugin, which would cause a
  dep cycle. Callers of SelectSessionWorkers must implement this func
  which is responsible for returning a storage bucket's worker filter as
  well as its associated credential id.
- authorizeSessionWithWorkerFilter now receives an
  internal/server.(*Repository) object when called. This is transparent
  to SelectSessionWorkers callers as it is handled internally.

* feat(server): Add FilteredWithFeatures fn and new server options

* refact(server): create new list and lookup worker queries

* test(server): Add WorkerList unit tests

* feat(server): Implement WorkerList randomness using crypto/rand pkg

* chore(db): add index to server_worker table

* refact(server): prefix test method with Test

* refact(server): rename Downstreamers interface to Graph

* chore(server): unexport and rename method

* chore: add doc string to WorkerAddress (#5420)

---------

Co-authored-by: Irena Rindos <irenarindos@users.noreply.github.com>
Co-authored-by: irenarindos <irena.rindos@gmail.com>

internal/server/options.go

@@ -23,37 +23,46 @@ func GetOpts(opt ...Option) options {
 // Option - how Options are passed as arguments
 type Option func(*options)
 
+// StorageBucketCredentialInfo defines the parameters to pass into the
+// WithFilterWorkersByStorageBucketCredentialState option.
+type StorageBucketCredentialInfo struct {
+	CredentialId string
+	Filters      []FilterStorageBucketCredentialStateFn
+}
+
 // options = how options are represented
 type options struct {
-	withName                               string
-	withPublicId                           string
-	withDescription                        string
-	withAddress                            string
-	withLimit                              int
-	withLiveness                           time.Duration
-	withUpdateTags                         bool
-	withWorkerTags                         []*Tag
-	withWorkerKeyIdentifier                string
-	withWorkerKeys                         WorkerKeys
-	withControllerEncryptionPrivateKey     []byte
-	withKeyId                              string
-	withNonce                              []byte
-	withNewIdFunc                          func(context.Context) (string, error)
-	WithFetchNodeCredentialsRequest        *types.FetchNodeCredentialsRequest
-	withTestPkiWorkerAuthorized            bool
-	withTestPkiWorkerKeyId                 *string
-	withTestUseInputTagsAsApiTags          bool
-	withWorkerType                         WorkerType
-	withRoot                               RootInfo
-	withStopAfter                          uint
-	WithCreateControllerLedActivationToken bool
-	withReleaseVersion                     string
-	withOperationalState                   string
-	withLocalStorageState                  string
-	withActiveWorkers                      bool
-	withFeature                            version.Feature
-	withDirectlyConnected                  bool
-	withWorkerPool                         []string
+	withName                                        string
+	withPublicId                                    string
+	withDescription                                 string
+	withAddress                                     string
+	withLimit                                       int
+	withLiveness                                    time.Duration
+	withUpdateTags                                  bool
+	withWorkerTags                                  []*Tag
+	withWorkerKeyIdentifier                         string
+	withWorkerKeys                                  WorkerKeys
+	withControllerEncryptionPrivateKey              []byte
+	withKeyId                                       string
+	withNonce                                       []byte
+	withNewIdFunc                                   func(context.Context) (string, error)
+	WithFetchNodeCredentialsRequest                 *types.FetchNodeCredentialsRequest
+	withTestPkiWorkerAuthorized                     bool
+	withTestPkiWorkerKeyId                          *string
+	withTestUseInputTagsAsApiTags                   bool
+	withWorkerType                                  WorkerType
+	withRoot                                        RootInfo
+	withStopAfter                                   uint
+	WithCreateControllerLedActivationToken          bool
+	withReleaseVersion                              string
+	withOperationalState                            string
+	withLocalStorageState                           string
+	withActiveWorkers                               bool
+	withFeature                                     version.Feature
+	withDirectlyConnected                           bool
+	withWorkerPool                                  []string
+	withFilterWorkersByStorageBucketCredentialState *StorageBucketCredentialInfo
+	withFilterWorkersByLocalStorageState            bool
 }
 
 func getDefaultOptions() options {
@@ -276,3 +285,21 @@ func WithLocalStorageState(state string) Option {
 		o.withLocalStorageState = state
 	}
 }
+
+// WithFilterWorkersByStorageBucketCredentialState receives a storage bucket
+// credential id and filters to apply and calls
+// FilterWorkersByStorageBucketCredentialState in supported repository
+// functions.
+func WithFilterWorkersByStorageBucketCredentialState(ci *StorageBucketCredentialInfo) Option {
+	return func(o *options) {
+		o.withFilterWorkersByStorageBucketCredentialState = ci
+	}
+}
+
+// WithFilterWorkersByLocalStorageState controls whether
+// FilterWorkersByLocalStorageState is called in supported repository functions.
+func WithFilterWorkersByLocalStorageState(filter bool) Option {
+	return func(o *options) {
+		o.withFilterWorkersByLocalStorageState = filter
+	}
+}

internal/server/query.go

@@ -4,6 +4,82 @@
 package server
 
 const (
+	listWorkersQuery = `
+	with connection_count (worker_id, count) as (
+    select worker_id,
+           count(1) as count
+      from session_connection
+     where closed_reason is null
+  group by worker_id
+  )
+    select w.public_id,
+           w.scope_id,
+           w.description,
+           w.name,
+           w.address,
+           w.create_time,
+           w.update_time,
+           w.version,
+           w.last_status_time,
+           w.type,
+           w.release_version,
+           w.operational_state,
+           w.local_storage_state,
+           cc.count as active_connection_count,
+           wt.tags as api_tags,
+           ct.tags as config_tags
+      from server_worker w
+ left join (   select worker_id, 
+                     json_agg(json_build_object('key', key, 'value', value)) as tags 
+                from server_worker_api_tag 
+               group by worker_id) wt
+        on w.public_id = wt.worker_id
+ left join (   select worker_id, 
+                     json_agg(json_build_object('key', key, 'value', value)) as tags 
+                from server_worker_config_tag group by worker_id) ct
+        on w.public_id = ct.worker_id
+ left join connection_count as cc
+        on w.public_id = cc.worker_id
+    `
+
+	lookupWorkerQuery = `
+	with connection_count (worker_id, count) as (
+    select worker_id,
+           count(1) as count
+      from session_connection
+     where closed_reason is null
+  group by worker_id
+  )
+    select w.public_id,
+           w.scope_id,
+           w.description,
+           w.name,
+           w.address,
+           w.create_time,
+           w.update_time,
+           w.version,
+           w.last_status_time,
+           w.type,
+           w.release_version,
+           w.operational_state,
+           w.local_storage_state,
+           cc.count as active_connection_count,
+           wt.tags as api_tags,
+           ct.tags as config_tags
+      from server_worker w
+ left join (   select worker_id, 
+                     json_agg(json_build_object('key', key, 'value', value)) as tags 
+                from server_worker_api_tag 
+               group by worker_id) wt
+        on w.public_id = wt.worker_id
+ left join (   select worker_id, 
+                     json_agg(json_build_object('key', key, 'value', value)) as tags 
+                from server_worker_config_tag group by worker_id) ct
+        on w.public_id = ct.worker_id
+ left join connection_count as cc
+        on w.public_id = cc.worker_id
+     where w.public_id = @worker_id
+    `
 	getStorageBucketCredentialStatesByWorkerId = `
 		select spsb.public_id as storage_bucket_id,
 			   wsbcs.permission_type, wsbcs.state, 
@@ -16,13 +92,17 @@ const (
 
 	deleteWhereCreateTimeSql = `create_time < ?`
 
-	deleteTagsByWorkerIdSql = `
+	deleteApiTagsByWorkerIdSql = `
 	delete 
-	from server_worker_tag 
-	where 
-		source = ?
-	and
-		worker_id = ?`
+	  from server_worker_api_tag
+	 where worker_id = ?
+	`
+
+	deleteConfigTagsByWorkerIdSql = `
+	delete 
+	  from server_worker_config_tag
+	 where worker_id = ?
+	`
 
 	deleteWorkerAuthQuery = `
 		delete from worker_auth_authorized
@@ -115,4 +195,32 @@ const (
 		where worker.scope_id = ?
 			and auth_token.key_id = ?
 	`
+
+	listHcpbManagedWorkersQuery = `
+        select w.public_id,
+               w.address
+          from server_worker            as w
+     left join server_worker_config_tag as t
+            on w.public_id = t.worker_id
+         where last_status_time > now() - interval '%d seconds'
+           and t.key   = 'boundary.cloud.hashicorp.com:managed'
+           and t.value = 'true';
+`
+
+	listSelectSessionWorkers = `
+        select w.public_id,
+               w.name,
+               w.address,
+               w.release_version,
+               w.local_storage_state,
+               wt.tags as api_tags,
+               ct.tags as config_tags
+          from server_worker w
+     left join (select worker_id, json_agg(json_build_object('key', key, 'value', value)) as tags from server_worker_api_tag group by worker_id) wt
+            on w.public_id = wt.worker_id
+     left join (select worker_id, json_agg(json_build_object('key', key, 'value', value)) as tags from server_worker_config_tag group by worker_id) ct
+            on w.public_id = ct.worker_id
+         where last_status_time > now() - interval '%d seconds'
+           and operational_state = 'active';
+`
 )

internal/server/repository_worker.go

@@ -5,15 +5,21 @@ package server
 
 import (
 	"context"
+	"database/sql"
 	stderrors "errors"
 	"fmt"
 	"strings"
+	"time"
 
+	"github.com/hashicorp/boundary/internal/daemon/controller/downstream"
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/globals"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/server/store"
+	"github.com/hashicorp/boundary/internal/target"
 	"github.com/hashicorp/boundary/internal/types/scope"
+	"github.com/hashicorp/go-bexpr"
 	"github.com/hashicorp/go-dbw"
 	"github.com/hashicorp/nodeenrollment"
 	"github.com/hashicorp/nodeenrollment/registration"
@@ -23,7 +29,21 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
-var ErrCannotUpdateKmsWorkerViaApi = stderrors.New("cannot update a kms worker's basic information via api")
+var (
+	ErrCannotUpdateKmsWorkerViaApi = stderrors.New("cannot update a kms worker's basic information via api")
+
+	FilterWorkersFn = filterWorkers
+)
+
+// StorageBucketFilterCredIdFn is a function that gets and returns a storage
+// bucket's worker filter and credential id.
+type StorageBucketFilterCredIdFn func(ctx context.Context, ce globals.ControllerExtension, sbId string) (filter string, credId string, err error)
+
+// WorkerAddress contains a worker's public id and address.
+type WorkerAddress struct {
+	PublicId string
+	Address  string
+}
 
 // DeleteWorker will delete a worker from the repository.
 func (r *Repository) DeleteWorker(ctx context.Context, publicId string, _ ...Option) (int, error) {
@@ -65,52 +85,6 @@ func (r *Repository) DeleteWorker(ctx context.Context, publicId string, _ ...Opt
 	return rowsDeleted, nil
 }
 
-// LookupWorkerByName returns the worker with the provided name. In the event
-// that no worker is found that matches then nil, nil will be returned.
-func (r *Repository) LookupWorkerByName(ctx context.Context, name string) (*Worker, error) {
-	const op = "server.(Repository).LookupWorkerByName"
-	switch {
-	case name == "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "name is empty")
-	}
-	w, err := lookupWorkerByName(ctx, r.reader, name)
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
-	}
-	if w == nil {
-		return nil, nil
-	}
-	w.RemoteStorageStates, err = r.ListWorkerStorageBucketCredentialState(ctx, w.GetPublicId())
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
-	}
-	return w, nil
-}
-
-func lookupWorkerByName(ctx context.Context, reader db.Reader, name string) (*Worker, error) {
-	const op = "server.lookupWorkerByName"
-	switch {
-	case isNil(reader):
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "reader is nil")
-	case name == "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "name is empty")
-	}
-
-	wAgg := &workerAggregate{}
-	err := reader.LookupWhere(ctx, &wAgg, "name = ?", []any{name})
-	if err != nil {
-		if errors.IsNotFoundError(err) {
-			return nil, nil
-		}
-		return nil, errors.Wrap(ctx, err, op)
-	}
-	w, err := wAgg.toWorker(ctx)
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
-	}
-	return w, nil
-}
-
 func (r *Repository) LookupWorkerIdByKeyId(ctx context.Context, keyId string) (string, error) {
 	const op = "server.(Repository).LookupWorkerIdByKeyId"
 	switch {
@@ -163,32 +137,63 @@ func lookupWorker(ctx context.Context, reader db.Reader, id string) (*Worker, er
 	case id == "":
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "id is empty")
 	}
-	wAgg := &workerAggregate{}
-	wAgg.PublicId = id
-	err := reader.LookupById(ctx, wAgg)
+
+	rows, err := reader.Query(ctx, lookupWorkerQuery, []any{sql.Named("worker_id", id)})
 	if err != nil {
-		if errors.IsNotFoundError(err) {
-			return nil, nil
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	defer rows.Close()
+	var worker *Worker
+	for rows.Next() {
+		if err := reader.ScanRows(ctx, rows, &worker); err != nil {
+			return nil, err
 		}
-		return nil, errors.Wrap(ctx, err, op)
 	}
-	w, err := wAgg.toWorker(ctx)
+	return worker, nil
+}
+
+// ListHcpbManagedWorkers lists all HCPb managed workers' ids and addresses.
+func (r *Repository) ListHcpbManagedWorkers(ctx context.Context, liveness time.Duration) ([]WorkerAddress, error) {
+	const op = "server.(Repository).ListHcpbManagedWorkers"
+
+	if liveness <= 0 {
+		liveness = DefaultLiveness
+	}
+	liveness = liveness.Truncate(time.Second)
+
+	query := fmt.Sprintf(listHcpbManagedWorkersQuery, uint32(liveness.Seconds()))
+	rows, err := r.reader.Query(ctx, query, []any{})
 	if err != nil {
 		return nil, errors.Wrap(ctx, err, op)
 	}
-	return w, nil
+	defer rows.Close()
+
+	var ret []WorkerAddress
+	for rows.Next() {
+		var result WorkerAddress
+		err = r.reader.ScanRows(ctx, rows, &result)
+		if err != nil {
+			return nil, errors.Wrap(ctx, err, op)
+		}
+		ret = append(ret, result)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+
+	return ret, nil
 }
 
 // ListWorkers will return a listing of Workers and honor the WithLimit option.
-// Supported options: WithWorkerType, WithActiveWorkers, WithLiveness,
-// WithWorkerPool, WithLimit
+// Supported options: WithLimit
 //
-// If WithLiveness is zero the default liveness value is used, if it is negative
-// then the last status update time is ignored.
 // If WithLimit < 0, then unlimited results are returned. If WithLimit == 0, then
-// default limits are used for results.  WithWorkerPool can be provided with a
-// non-zero length slice of worker ids to restrict the returned workers to only
-// ones with the ids provided.
+// default limits are used for results.
+//
+// Please note the cost of calling ListWorkers, as the underlying query calculates a connection count
+// for each worker by joining with the session connection table. Instead of using this function to get
+// worker information, create a domain-specific function that only returns the information you need.
+// The purpose of ListWorkers is to return a list of workers for the API.
 func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ...Option) ([]*Worker, error) {
 	const op = "server.(Repository).ListWorkers"
 
@@ -201,30 +206,20 @@ func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ...
 	default:
 		newOpts = append(newOpts, WithLimit(r.defaultLimit))
 	}
-	// handle the WithLiveness default
-	switch {
-	case opts.withLiveness != 0:
-		newOpts = append(newOpts, WithLiveness(opts.withLiveness))
-	default:
-		newOpts = append(newOpts, WithLiveness(DefaultLiveness))
 
-	}
-	newOpts = append(newOpts, WithWorkerType(opts.withWorkerType))
-	newOpts = append(newOpts, WithActiveWorkers(opts.withActiveWorkers))
-	newOpts = append(newOpts, WithWorkerPool(opts.withWorkerPool))
 	return ListWorkers(ctx, r.reader, scopeIds, newOpts...)
 }
 
 // ListWorkers will return a listing of Workers and honor the WithLimit option.
-// Supported options: WithWorkerType, WithActiveWorkers, WithLiveness,
-// WithWorkerPool, WithLimit
+// Supported options: WithLimit
 //
-// If WithLiveness is zero the default liveness value is used, if it is negative
-// then the last status update time is ignored.
 // If WithLimit < 0, then unlimited results are returned. If WithLimit == 0, then
-// default limits are used for results.  WithWorkerPool can be provided with a
-// non-zero length slice of worker ids to restrict the returned workers to only
-// ones with the ids provided.
+// default limits are used for results.
+//
+// Please note the cost of calling ListWorkers, as the underlying query calculates a connection count
+// for each worker by joining with the session connection table. Instead of using this function to get
+// worker information, create a domain-specific function that only returns the information you need.
+// The purpose of ListWorkers is to return a list of workers for the API.
 func ListWorkers(ctx context.Context, reader db.Reader, scopeIds []string, opt ...Option) ([]*Worker, error) {
 	const op = "server.ListWorkers"
 	switch {
@@ -233,65 +228,39 @@ func ListWorkers(ctx context.Context, reader db.Reader, scopeIds []string, opt .
 	}
 
 	opts := GetOpts(opt...)
-	liveness := opts.withLiveness
-	if liveness == 0 {
-		liveness = DefaultLiveness
-	}
 
-	var where []string
+	query := listWorkersQuery
 	var whereArgs []any
-	if liveness > 0 {
-		where = append(where, fmt.Sprintf("last_status_time > now() - interval '%d seconds'", uint32(liveness.Seconds())))
-	}
+
 	if len(scopeIds) > 0 {
-		where = append(where, "scope_id in (?)")
+		query = fmt.Sprintf("%s where scope_id in (?)", query)
 		whereArgs = append(whereArgs, scopeIds)
 	}
 
-	switch opts.withWorkerType {
-	case "":
-	case KmsWorkerType, PkiWorkerType:
-		where = append(where, "type = ?")
-		whereArgs = append(whereArgs, opts.withWorkerType.String())
-	default:
-		return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown worker type %v", opts.withWorkerType))
-	}
-
-	if opts.withActiveWorkers {
-		where = append(where, "operational_state = ?")
-		whereArgs = append(whereArgs, ActiveOperationalState.String())
-	}
-
-	if len(opts.withWorkerPool) > 0 {
-		where = append(where, "public_id in (?)")
-		whereArgs = append(whereArgs, opts.withWorkerPool)
-	}
-
 	limit := db.DefaultLimit
 	if opts.withLimit != 0 {
 		// non-zero signals an override of the default limit for the repo.
 		limit = opts.withLimit
 	}
 
-	var wAggs []*workerAggregate
-	if err := reader.SearchWhere(
-		ctx,
-		&wAggs,
-		strings.Join(where, " and "),
-		whereArgs,
-		db.WithLimit(limit),
-	); err != nil {
-		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error searching for workers"))
+	if limit > 0 {
+		query = fmt.Sprintf("%s limit %d", query, limit)
 	}
 
-	workers := make([]*Worker, 0, len(wAggs))
-	for _, a := range wAggs {
-		w, err := a.toWorker(ctx)
-		if err != nil {
-			return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error converting workerAggregate to Worker"))
-		}
-		workers = append(workers, w)
+	var workers []*Worker
+	rows, err := reader.Query(ctx, query, whereArgs)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error searching for workers"))
 	}
+	defer rows.Close()
+	for rows.Next() {
+		var worker Worker
+		if err := reader.ScanRows(context.Background(), rows, &worker); err != nil {
+			return nil, err
+		}
+		workers = append(workers, &worker)
+	}
+
 	return workers, nil
 }
 
@@ -302,25 +271,25 @@ func ListWorkers(ctx context.Context, reader db.Reader, scopeIds []string, opt .
 // The WithPublicId, WithKeyId, and WithUpdateTags options are
 // the only ones used. All others are ignored.
 // Workers are intentionally not oplogged.
-func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt ...Option) (*Worker, error) {
+func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt ...Option) (string, error) {
 	const op = "server.(Repository).UpsertWorkerStatus"
 
 	opts := GetOpts(opt...)
 	switch {
 	case worker == nil:
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker is nil")
+		return "", errors.New(ctx, errors.InvalidParameter, op, "worker is nil")
 	case worker.GetAddress() == "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker reported address is empty")
+		return "", errors.New(ctx, errors.InvalidParameter, op, "worker reported address is empty")
 	case worker.ScopeId == "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "scope id is empty")
+		return "", errors.New(ctx, errors.InvalidParameter, op, "scope id is empty")
 	case worker.PublicId != "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker id is not empty")
+		return "", errors.New(ctx, errors.InvalidParameter, op, "worker id is not empty")
 	case worker.GetName() == "" && opts.withKeyId == "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both empty; one is required")
+		return "", errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both empty; one is required")
 	case worker.OperationalState == "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker operational state is empty")
+		return "", errors.New(ctx, errors.InvalidParameter, op, "worker operational state is empty")
 	case worker.LocalStorageState == "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker local storage state is empty")
+		return "", errors.New(ctx, errors.InvalidParameter, op, "worker local storage state is empty")
 	}
 
 	var workerId string
@@ -331,7 +300,7 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt
 	case opts.withKeyId != "":
 		workerId, err = r.LookupWorkerIdByKeyId(ctx, opts.withKeyId)
 		if err != nil || workerId == "" {
-			return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error finding worker by keyId"))
+			return "", errors.Wrap(ctx, err, op, errors.WithMsg("error finding worker by keyId"))
 		}
 	default:
 		// generating the worker id based off of the scope and name ensures
@@ -341,11 +310,10 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt
 		// workers and kms workers.
 		workerId, err = NewWorkerIdFromScopeAndName(ctx, worker.GetScopeId(), worker.GetName())
 		if err != nil || workerId == "" {
-			return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker id"))
+			return "", errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker id"))
 		}
 	}
 
-	var ret *Worker
 	_, err = r.writer.DoTx(
 		ctx,
 		db.StdRetryCnt,
@@ -404,31 +372,22 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt
 
 			// If we've been told to update tags, we need to clean out old
 			// ones and add new ones. Within the current transaction, simply
-			// delete all tags for the given worker, then add the new ones
+			// delete all config tags for the given worker, then add the new ones
 			// we've been sent.
 			if opts.withUpdateTags {
-				if err := setWorkerTags(ctx, w, workerClone.GetPublicId(), ConfigurationTagSource, workerClone.inputTags); err != nil {
+				if err := setWorkerConfigTags(ctx, w, workerClone.GetPublicId(), workerClone.inputTags); err != nil {
 					return errors.Wrap(ctx, err, op, errors.WithMsg("error setting worker tags"))
 				}
 			}
 
-			wAgg := &workerAggregate{PublicId: workerClone.GetPublicId()}
-			if err := reader.LookupById(ctx, wAgg); err != nil {
-				return errors.Wrap(ctx, err, op, errors.WithMsg("error looking up worker aggregate"))
-			}
-			ret, err = wAgg.toWorker(ctx)
-			if err != nil {
-				return errors.Wrap(ctx, err, op, errors.WithMsg("error converting worker aggregate to worker"))
-			}
-
 			return nil
 		},
 	)
 	if err != nil {
-		return nil, err
+		return "", err
 	}
 
-	return ret, nil
+	return workerId, nil
 }
 
 // VerifyKnownWorkers checks that the passed worker IDs are found in the repository and returns
@@ -465,21 +424,19 @@ func (r *Repository) VerifyKnownWorkers(ctx context.Context, ids []string) ([]st
 	return ret, nil
 }
 
-// setWorkerTags removes all existing tags from the same source and worker id
+// setWorkerConfigTags removes all existing config tags from the same source and worker id
 // and creates new ones based on the ones provided.  This function should be
 // called from inside a db transaction.
 // Workers/worker tags are intentionally not oplogged.
-func setWorkerTags(ctx context.Context, w db.Writer, id string, ts TagSource, tags []*Tag) error {
-	const op = "server.setWorkerTags"
+func setWorkerConfigTags(ctx context.Context, w db.Writer, id string, tags []*Tag) error {
+	const op = "server.setWorkerConfigTags"
 	switch {
-	case !ts.isValid():
-		return errors.New(ctx, errors.InvalidParameter, op, "invalid tag source provided")
 	case id == "":
 		return errors.New(ctx, errors.InvalidParameter, op, "worker id is empty")
 	case isNil(w):
 		return errors.New(ctx, errors.InvalidParameter, op, "db.Writer is nil")
 	}
-	_, err := w.Exec(ctx, deleteTagsByWorkerIdSql, []any{ts.String(), id})
+	_, err := w.Exec(ctx, deleteConfigTagsByWorkerIdSql, []any{id})
 	if err != nil {
 		return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("couldn't delete existing tags for worker %q", id)))
 	}
@@ -489,16 +446,56 @@ func setWorkerTags(ctx context.Context, w db.Writer, id string, ts TagSource, ta
 	// Otherwise, go through and stage each tuple for insertion
 	// below.
 	if len(tags) > 0 {
-		uTags := make([]*store.WorkerTag, 0, len(tags))
+		uTags := make([]*store.ConfigTag, 0, len(tags))
 		for _, v := range tags {
 			if v == nil {
 				return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("found nil tag value for worker %s", id))
 			}
-			uTags = append(uTags, &store.WorkerTag{
+			uTags = append(uTags, &store.ConfigTag{
+				WorkerId: id,
+				Key:      v.Key,
+				Value:    v.Value,
+			})
+		}
+		if err = w.CreateItems(ctx, uTags); err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("error creating tags for worker %q", id)))
+		}
+	}
+
+	return nil
+}
+
+// setWorkerApiTags removes all existing API tags from the same source and worker id
+// and creates new ones based on the ones provided.  This function should be
+// called from inside a db transaction.
+// Workers/worker tags are intentionally not oplogged.
+func setWorkerApiTags(ctx context.Context, w db.Writer, id string, tags []*Tag) error {
+	const op = "server.setWorkerApiTags"
+	switch {
+	case id == "":
+		return errors.New(ctx, errors.InvalidParameter, op, "worker id is empty")
+	case isNil(w):
+		return errors.New(ctx, errors.InvalidParameter, op, "db.Writer is nil")
+	}
+	_, err := w.Exec(ctx, deleteApiTagsByWorkerIdSql, []any{id})
+	if err != nil {
+		return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("couldn't delete existing tags for worker %q", id)))
+	}
+
+	// If tags were cleared out entirely, then we'll have nothing
+	// to do here, e.g., it will result in deletion of all tags.
+	// Otherwise, go through and stage each tuple for insertion
+	// below.
+	if len(tags) > 0 {
+		uTags := make([]*store.ApiTag, 0, len(tags))
+		for _, v := range tags {
+			if v == nil {
+				return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("found nil tag value for worker %s", id))
+			}
+			uTags = append(uTags, &store.ApiTag{
 				WorkerId: id,
 				Key:      v.Key,
 				Value:    v.Value,
-				Source:   ts.String(),
 			})
 		}
 		if err = w.CreateItems(ctx, uTags); err != nil {
@@ -565,16 +562,19 @@ func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version u
 			// First we need to do a lookup so we can validate that this
 			// function is not being used for workers registered via KMS-PKI
 			// means
-			wAgg := &workerAggregate{PublicId: worker.GetPublicId()}
-			if err := reader.LookupById(ctx, wAgg); err != nil {
+			wLookup, err := lookupWorker(ctx, reader, worker.GetPublicId())
+			if err != nil {
 				return errors.Wrap(ctx, err, op)
 			}
+			if wLookup == nil {
+				return errors.New(ctx, errors.RecordNotFound, op, fmt.Sprintf("failed to find worker %q", worker.GetPublicId()))
+			}
 			// If it's a KMS-PKI worker we do not want to allow
 			// name/description/other updates via the API, it should only come
 			// in via the upsert mechanism via status updates. If the public ID
 			// is predictably generated in the KMS fashion, it's a KMS-PKI
 			// worker.
-			workerId, err := NewWorkerIdFromScopeAndName(ctx, wAgg.ScopeId, wAgg.Name)
+			workerId, err := NewWorkerIdFromScopeAndName(ctx, wLookup.ScopeId, wLookup.Name)
 			if err != nil {
 				return errors.Wrap(ctx, err, op, errors.WithMsg("error generating worker id in kms-pki name check case"))
 			}
@@ -596,13 +596,10 @@ func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version u
 				return errors.New(ctx, errors.MultipleRecords, op, "more than 1 resource would have been updated")
 			}
 
-			wAgg = &workerAggregate{PublicId: worker.GetPublicId()}
-			if err := reader.LookupById(ctx, wAgg); err != nil {
+			ret, err = lookupWorker(ctx, reader, worker.GetPublicId())
+			if err != nil {
 				return errors.Wrap(ctx, err, op)
 			}
-			if ret, err = wAgg.toWorker(ctx); err != nil {
-				return err
-			}
 			ret.RemoteStorageStates, err = r.ListWorkerStorageBucketCredentialState(ctx, ret.GetPublicId())
 			if err != nil {
 				return err
@@ -768,7 +765,7 @@ func (r *Repository) AddWorkerTags(ctx context.Context, workerId string, workerV
 		return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("no worker found with public id %s", workerId))
 	}
 
-	newTags := append(worker.apiTags, tags...)
+	newTags := append(worker.ApiTags, tags...)
 	_, err = r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error {
 		worker := worker.clone()
 		worker.PublicId = workerId
@@ -780,7 +777,7 @@ func (r *Repository) AddWorkerTags(ctx context.Context, workerId string, workerV
 		if rowsUpdated != 1 {
 			return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("updated worker version and %d rows updated", rowsUpdated))
 		}
-		err = setWorkerTags(ctx, w, workerId, ApiTagSource, newTags)
+		err = setWorkerApiTags(ctx, w, workerId, newTags)
 		if err != nil {
 			return errors.Wrap(ctx, err, op)
 		}
@@ -822,7 +819,7 @@ func (r *Repository) SetWorkerTags(ctx context.Context, workerId string, workerV
 		if rowsUpdated != 1 {
 			return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("updated worker version and %d rows updated", rowsUpdated))
 		}
-		err = setWorkerTags(ctx, w, workerId, ApiTagSource, tags)
+		err = setWorkerApiTags(ctx, w, workerId, tags)
 		if err != nil {
 			return errors.Wrap(ctx, err, op)
 		}
@@ -856,16 +853,15 @@ func (r *Repository) DeleteWorkerTags(ctx context.Context, workerId string, work
 	}
 
 	rowsDeleted := 0
-	deleteTags := make([]*store.WorkerTag, 0, len(tags))
+	deleteTags := make([]*store.ApiTag, 0, len(tags))
 	for _, t := range tags {
 		if t == nil {
 			return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "found nil tag value in input")
 		}
-		deleteTags = append(deleteTags, &store.WorkerTag{
+		deleteTags = append(deleteTags, &store.ApiTag{
 			WorkerId: workerId,
 			Key:      t.Key,
 			Value:    t.Value,
-			Source:   ApiTagSource.String(),
 		})
 	}
 
@@ -897,3 +893,113 @@ func (r *Repository) DeleteWorkerTags(ctx context.Context, workerId string, work
 	}
 	return rowsDeleted, nil
 }
+
+// SelectSessionWorkers attempts to select suitable workers for a given session.
+// Additionally, it returns the data for the the protocol-aware worker that can
+// handle recording operations, if enabled.
+func (r *Repository) SelectSessionWorkers(ctx context.Context,
+	workerStatusGracePeriod time.Duration,
+	t target.Target,
+	host string,
+	ce globals.ControllerExtension,
+	sbFn StorageBucketFilterCredIdFn,
+	ds downstream.Graph,
+) ([]WorkerAddress, string, error) {
+	const op = "server.(Repository).SelectSessionWorkers"
+
+	if workerStatusGracePeriod <= 0 {
+		workerStatusGracePeriod = DefaultLiveness
+	}
+	workerStatusGracePeriod = workerStatusGracePeriod.Truncate(time.Second)
+
+	query := fmt.Sprintf(listSelectSessionWorkers, uint32(workerStatusGracePeriod.Seconds()))
+	var livingWorkers []*Worker
+	_, err := r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			rows, err := r.reader.Query(ctx, query, []any{})
+			if err != nil {
+				return err
+			}
+			defer rows.Close()
+			for rows.Next() {
+				// Note: The Worker objects scanned here do not contain all data
+				// a Worker object can hold, only a subset. Check the query to
+				// learn exactly what fields are present.
+				var worker Worker
+				if err := r.reader.ScanRows(ctx, rows, &worker); err != nil {
+					return err
+				}
+				livingWorkers = append(livingWorkers, &worker)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return nil, "", errors.Wrap(ctx, err, op, errors.WithMsg("error searching for workers"))
+	}
+	if len(livingWorkers) == 0 {
+		return nil, "", errors.New(ctx, errors.WorkerNotFoundForRequest, op, "No workers are available to handle this session.")
+	}
+
+	wl, protoWorker, err := FilterWorkersFn(ctx, r, t, livingWorkers, host, ce, sbFn, ds)
+	if err != nil {
+		return nil, "", errors.Wrap(ctx, err, op)
+	}
+
+	was := make([]WorkerAddress, 0, len(wl))
+	for _, w := range wl {
+		was = append(was, WorkerAddress{PublicId: w.GetPublicId(), Address: w.GetAddress()})
+	}
+
+	if protoWorker != nil {
+		return was, protoWorker.GetPublicId(), nil
+	}
+	return was, "", nil
+}
+
+// If set, use the worker_filter or egress_worker_filter to filter the selected workers
+// and ensure we have workers available to service this request. The second return
+// argument is always nil.
+func filterWorkers(
+	ctx context.Context,
+	_ *Repository,
+	t target.Target,
+	selectedWorkers WorkerList,
+	_ string,
+	_ globals.ControllerExtension,
+	_ StorageBucketFilterCredIdFn,
+	_ downstream.Graph,
+	_ ...target.Option,
+) (WorkerList, *Worker, error) {
+	const op = "server.filterWorkers"
+
+	if len(selectedWorkers) > 0 {
+		var eval *bexpr.Evaluator
+		var err error
+		switch {
+		case len(t.GetEgressWorkerFilter()) > 0:
+			eval, err = bexpr.CreateEvaluator(t.GetEgressWorkerFilter())
+		case len(t.GetWorkerFilter()) > 0:
+			eval, err = bexpr.CreateEvaluator(t.GetWorkerFilter())
+		default: // No filter
+			return selectedWorkers, nil, nil
+		}
+		if err != nil {
+			return nil, nil, err
+		}
+
+		selectedWorkers, err = selectedWorkers.Filtered(eval)
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+
+	if len(selectedWorkers) == 0 {
+		return nil, nil, errors.New(ctx, errors.WorkerNotFoundForRequest, op, "No workers are available to handle this session, or all have been filtered.")
+	}
+
+	return selectedWorkers, nil, nil
+}

internal/server/repository_worker_test.go

@@ -22,6 +22,9 @@ import (
 	"github.com/hashicorp/boundary/internal/server"
 	"github.com/hashicorp/boundary/internal/server/store"
 	"github.com/hashicorp/boundary/internal/session"
+	"github.com/hashicorp/boundary/internal/target"
+	"github.com/hashicorp/boundary/internal/target/targettest"
+	tgstore "github.com/hashicorp/boundary/internal/target/targettest/store"
 	"github.com/hashicorp/boundary/internal/types/scope"
 	"github.com/hashicorp/go-dbw"
 	"github.com/hashicorp/go-kms-wrapping/extras/kms/v2/migrations"
@@ -110,39 +113,6 @@ func TestDeleteWorker(t *testing.T) {
 	}
 }
 
-func TestLookupWorkerByName(t *testing.T) {
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	repo, err := server.NewRepository(ctx, rw, rw, kms)
-	require.NoError(t, err)
-
-	w := server.TestKmsWorker(t, conn, wrapper)
-	t.Run("success", func(t *testing.T) {
-		got, err := repo.LookupWorkerByName(ctx, w.GetName())
-		require.NoError(t, err)
-		assert.Empty(t, cmp.Diff(w.Worker, got.Worker, protocmp.Transform()))
-	})
-	t.Run("not found", func(t *testing.T) {
-		got, err := repo.LookupWorkerByName(ctx, "unknown_name")
-		require.NoError(t, err)
-		assert.Nil(t, got)
-	})
-	t.Run("db error", func(t *testing.T) {
-		conn, mock := db.TestSetupWithMock(t)
-		rw := db.New(conn)
-		mock.ExpectQuery(`SELECT`).WillReturnError(errors.New(ctx, errors.Internal, "test", "lookup-error"))
-		r, err := server.NewRepository(ctx, rw, rw, kms)
-		require.NoError(t, err)
-		got, err := r.LookupWorkerByName(ctx, w.GetName())
-		assert.NoError(t, mock.ExpectationsWereMet())
-		assert.Truef(t, errors.Match(errors.T(errors.Op("server.(Repository).LookupWorkerByName")), err), "got error %v", err)
-		assert.Nil(t, got)
-	})
-}
-
 func TestLookupWorkerIdByKeyId(t *testing.T) {
 	ctx := context.Background()
 	conn, _ := db.TestSetup(t, "postgres")
@@ -229,7 +199,7 @@ func TestLookupWorker(t *testing.T) {
 		got, err := repo.LookupWorker(ctx, w.GetPublicId())
 		require.NoError(t, err)
 		assert.Empty(t, cmp.Diff(w, got, protocmp.Transform()))
-		assert.Equal(t, uint32(3), got.ActiveConnectionCount())
+		assert.Equal(t, uint32(3), got.ActiveConnectionCount)
 		assert.Equal(t, map[string][]string{
 			"key": {"val"},
 		}, got.CanonicalTags())
@@ -243,7 +213,7 @@ func TestLookupWorker(t *testing.T) {
 	t.Run("db error", func(t *testing.T) {
 		conn, mock := db.TestSetupWithMock(t)
 		rw := db.New(conn)
-		mock.ExpectQuery(`SELECT`).WillReturnError(errors.New(context.Background(), errors.Internal, "test", "lookup-error"))
+		mock.ExpectQuery(`with connection_count`).WillReturnError(errors.New(context.Background(), errors.Internal, "test", "lookup-error"))
 		r, err := server.NewRepository(ctx, rw, rw, kms)
 		require.NoError(t, err)
 		got, err := r.LookupWorker(ctx, w.GetPublicId())
@@ -268,7 +238,7 @@ func TestUpsertWorkerStatus(t *testing.T) {
 			server.WithAddress("address"), server.WithName("config_name1"),
 			server.WithDescription("kms_description1"),
 		)
-		worker, err := repo.UpsertWorkerStatus(ctx, wStatus1)
+		worker, err := server.TestUpsertAndReturnWorker(ctx, t, wStatus1, repo)
 		require.NoError(t, err)
 
 		assert.True(t, strings.HasPrefix(worker.GetPublicId(), "w_"))
@@ -283,7 +253,7 @@ func TestUpsertWorkerStatus(t *testing.T) {
 		// update again and see updated last status time
 		wStatus2 := server.NewWorker(scope.Global.String(),
 			server.WithAddress("new_address"), server.WithName("config_name1"), server.WithReleaseVersion("test-version"))
-		worker, err = repo.UpsertWorkerStatus(ctx, wStatus2)
+		worker, err = server.TestUpsertAndReturnWorker(ctx, t, wStatus2, repo)
 		require.NoError(t, err)
 		assert.Greater(t, worker.GetLastStatusTime().AsTime(), worker.GetCreateTime().AsTime())
 		assert.Equal(t, "config_name1", worker.Name)
@@ -293,7 +263,7 @@ func TestUpsertWorkerStatus(t *testing.T) {
 		assert.Equal(t, "new_address", worker.GetAddress())
 
 		// Expect this worker to be returned as it is active
-		workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithActiveWorkers(true))
+		workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()})
 		require.NoError(t, err)
 		assert.Len(t, workers, 1)
 		assert.Equal(t, server.KmsWorkerType.String(), workers[0].Type)
@@ -302,7 +272,7 @@ func TestUpsertWorkerStatus(t *testing.T) {
 		wStatus3 := server.NewWorker(scope.Global.String(),
 			server.WithAddress("new_address"), server.WithName("config_name1"),
 			server.WithOperationalState("shutdown"), server.WithReleaseVersion("Boundary v0.11.0"))
-		worker, err = repo.UpsertWorkerStatus(ctx, wStatus3)
+		worker, err = server.TestUpsertAndReturnWorker(ctx, t, wStatus3, repo)
 		require.NoError(t, err)
 		assert.Greater(t, worker.GetLastStatusTime().AsTime(), worker.GetCreateTime().AsTime())
 		// Version does not change for status updates
@@ -315,17 +285,17 @@ func TestUpsertWorkerStatus(t *testing.T) {
 			server.WithAddress("new_address"), server.WithName("config_name1"),
 			server.WithOperationalState("shutdown"), server.WithReleaseVersion("Boundary v0.11.0"),
 			server.WithLocalStorageState("available"))
-		worker, err = repo.UpsertWorkerStatus(ctx, wStatus4)
+		worker, err = server.TestUpsertAndReturnWorker(ctx, t, wStatus4, repo)
 		require.NoError(t, err)
 		assert.Greater(t, worker.GetLastStatusTime().AsTime(), worker.GetCreateTime().AsTime())
 		// Version does not change for status updates
 		assert.Equal(t, uint32(1), worker.Version)
 		assert.Equal(t, server.AvailableLocalStorageState.String(), worker.GetLocalStorageState())
 
-		// Should no longer see this worker in listing if we exclude shutdown workers
-		workers, err = repo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithActiveWorkers(true))
+		// Check that this worker now returns in the shutdown state
+		workers, err = repo.ListWorkers(ctx, []string{scope.Global.String()})
 		require.NoError(t, err)
-		assert.Len(t, workers, 0)
+		assert.Equal(t, "shutdown", workers[0].GetOperationalState())
 	})
 
 	// Setup and use a pki worker
@@ -336,7 +306,7 @@ func TestUpsertWorkerStatus(t *testing.T) {
 		wStatus1 := server.NewWorker(scope.Global.String(),
 			server.WithAddress("pki_address"), server.WithDescription("pki_description2"),
 			server.WithReleaseVersion("test-version"))
-		worker, err := repo.UpsertWorkerStatus(ctx, wStatus1, server.WithKeyId(pkiWorkerKeyId), server.WithReleaseVersion("test-version"))
+		worker, err := server.TestUpsertAndReturnWorker(ctx, t, wStatus1, repo, server.WithKeyId(pkiWorkerKeyId), server.WithReleaseVersion("test-version"))
 		require.NoError(t, err)
 
 		assert.True(t, strings.HasPrefix(worker.GetPublicId(), "w_"))
@@ -494,10 +464,17 @@ func TestUpsertWorkerStatus(t *testing.T) {
 		_, err = repo.UpsertWorkerStatus(ctx, anotherStatus)
 		require.NoError(t, err)
 
-		// Filtering out shutdown workers will remove the shutdown KMS and this shutdown worker, resulting in 2
-		workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithActiveWorkers(true))
+		// Ensure that we find two shutdown workers
+		workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()})
 		require.NoError(t, err)
-		assert.Len(t, workers, 2)
+		assert.Len(t, workers, 4)
+		numShutdown := 0
+		for _, w := range workers {
+			if w.OperationalState == "shutdown" {
+				numShutdown++
+			}
+		}
+		assert.Equal(t, 2, numShutdown)
 	})
 
 	t.Run("ipv4 address", func(t *testing.T) {
@@ -648,8 +625,7 @@ func TestTagUpdatingListing(t *testing.T) {
 				Value: "value2",
 			}))
 
-	worker1, err = repo.UpsertWorkerStatus(ctx, wStatus,
-		server.WithUpdateTags(true))
+	worker1, err = server.TestUpsertAndReturnWorker(ctx, t, wStatus, repo, server.WithUpdateTags(true))
 	require.NoError(err)
 	assert.Len(t, worker1.CanonicalTags(), 1)
 	assert.ElementsMatch(t, []string{"value1", "value2"}, worker1.CanonicalTags()["tag1"])
@@ -667,13 +643,13 @@ func TestTagUpdatingListing(t *testing.T) {
 				Key:   "tag22",
 				Value: "value22",
 			}))
-	worker1, err = repo.UpsertWorkerStatus(ctx, wStatus)
+	worker1, err = server.TestUpsertAndReturnWorker(ctx, t, wStatus, repo)
 	require.NoError(err)
 	assert.Len(t, worker1.CanonicalTags(), 1)
 	assert.ElementsMatch(t, []string{"value1", "value2"}, worker1.CanonicalTags()["tag1"])
 
 	// Update tags and test again
-	worker1, err = repo.UpsertWorkerStatus(ctx, wStatus, server.WithUpdateTags(true))
+	worker1, err = server.TestUpsertAndReturnWorker(ctx, t, wStatus, repo, server.WithUpdateTags(true))
 	require.NoError(err)
 	assert.Len(t, worker1.CanonicalTags(), 1)
 	assert.ElementsMatch(t, []string{"value21", "value22"}, worker1.CanonicalTags()["tag22"])
@@ -708,29 +684,6 @@ func TestListWorkers(t *testing.T) {
 			wantCnt:   testLimit*2 + 2,
 			wantErr:   false,
 		},
-		{
-			name:      "only-kms-type",
-			createCnt: testLimit + 1,
-			reqScopes: []string{scope.Global.String()},
-			opts:      []server.Option{server.WithLimit(-1), server.WithWorkerType(server.KmsWorkerType)},
-			wantCnt:   testLimit + 1,
-			wantErr:   false,
-		},
-		{
-			name:      "only-pki-type",
-			createCnt: testLimit + 1,
-			reqScopes: []string{scope.Global.String()},
-			opts:      []server.Option{server.WithLimit(-1), server.WithWorkerType(server.PkiWorkerType)},
-			wantCnt:   testLimit + 1,
-			wantErr:   false,
-		},
-		{
-			name:      "bad-type",
-			createCnt: testLimit + 1,
-			reqScopes: []string{scope.Global.String()},
-			opts:      []server.Option{server.WithLimit(-1), server.WithWorkerType("foo")},
-			wantErr:   true,
-		},
 		{
 			name:      "default-limit",
 			createCnt: testLimit + 1,
@@ -763,10 +716,7 @@ func TestListWorkers(t *testing.T) {
 			for i := 0; i < tt.createCnt; i++ {
 				server.TestPkiWorker(t, conn, wrapper)
 			}
-			// the purpose of these tests isn't to check liveness, so disable
-			// liveness checking.
-			opts := append(tt.opts, server.WithLiveness(-1))
-			got, err := repo.ListWorkers(ctx, tt.reqScopes, opts...)
+			got, err := repo.ListWorkers(ctx, tt.reqScopes, tt.opts...)
 			if tt.wantErr {
 				assert.Error(t, err)
 				return
@@ -788,253 +738,6 @@ func TestListWorkers(t *testing.T) {
 	}
 }
 
-func TestListWorkers_WithWorkerPool(t *testing.T) {
-	t.Parallel()
-	ctx := context.Background()
-	require := require.New(t)
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kmsCache := kms.TestKms(t, conn, wrapper)
-	require.NoError(kmsCache.CreateKeys(context.Background(), scope.Global.String(), kms.WithRandomReader(rand.Reader)))
-
-	serversRepo, err := server.NewRepository(ctx, rw, rw, kmsCache)
-	require.NoError(err)
-
-	worker1 := server.TestKmsWorker(t, conn, wrapper)
-	worker2 := server.TestPkiWorker(t, conn, wrapper)
-	worker3 := server.TestKmsWorker(t, conn, wrapper)
-
-	result, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithLiveness(-1))
-	require.NoError(err)
-	require.Len(result, 3)
-
-	tests := []struct {
-		name       string
-		workerPool []string
-		want       []*server.Worker
-	}{
-		{
-			name:       "one",
-			workerPool: []string{worker1.GetPublicId()},
-			want:       []*server.Worker{worker1},
-		},
-		{
-			name:       "two",
-			workerPool: []string{worker2.GetPublicId(), worker3.GetPublicId()},
-			want:       []*server.Worker{worker2, worker3},
-		},
-		{
-			name:       "none",
-			workerPool: []string{},
-			want:       []*server.Worker{worker1, worker2, worker3},
-		},
-		{
-			name:       "unknown ids",
-			workerPool: []string{"this_is_unknown"},
-			want:       []*server.Worker{},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithLiveness(-1), server.WithWorkerPool(tt.workerPool))
-			require.NoError(err)
-			assert.ElementsMatch(t, tt.want, got)
-		})
-	}
-}
-
-func TestListWorkers_WithActiveWorkers(t *testing.T) {
-	t.Parallel()
-	ctx := context.Background()
-	require := require.New(t)
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	serversRepo, err := server.NewRepository(ctx, rw, rw, kms)
-	require.NoError(err)
-
-	worker1 := server.TestKmsWorker(t, conn, wrapper)
-	worker2 := server.TestKmsWorker(t, conn, wrapper)
-	worker3 := server.TestKmsWorker(t, conn, wrapper)
-
-	result, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()})
-	require.NoError(err)
-	require.Len(result, 3)
-
-	tests := []struct {
-		name      string
-		upsertFn  func() (*server.Worker, error)
-		wantCnt   int
-		wantState string
-	}{
-		{
-			name: "upsert-worker1-to-shutdown",
-			upsertFn: func() (*server.Worker, error) {
-				return serversRepo.UpsertWorkerStatus(ctx,
-					server.NewWorker(scope.Global.String(),
-						server.WithName(worker1.GetName()),
-						server.WithAddress(worker1.GetAddress()),
-						server.WithOperationalState(server.ShutdownOperationalState.String()),
-						server.WithReleaseVersion("Boundary v.0.11"),
-						server.WithPublicId(worker1.GetPublicId())))
-			},
-			wantCnt:   2,
-			wantState: server.ShutdownOperationalState.String(),
-		},
-		{
-			name: "upsert-worker2-to-shutdown",
-			upsertFn: func() (*server.Worker, error) {
-				return serversRepo.UpsertWorkerStatus(ctx,
-					server.NewWorker(scope.Global.String(),
-						server.WithName(worker2.GetName()),
-						server.WithAddress(worker2.GetAddress()),
-						server.WithOperationalState(server.ShutdownOperationalState.String()),
-						server.WithReleaseVersion("Boundary v.0.11"),
-						server.WithPublicId(worker2.GetPublicId())))
-			},
-			wantCnt:   1,
-			wantState: server.ShutdownOperationalState.String(),
-		},
-		{
-			name: "upsert-worker3-to-shutdown",
-			upsertFn: func() (*server.Worker, error) {
-				return serversRepo.UpsertWorkerStatus(ctx,
-					server.NewWorker(scope.Global.String(),
-						server.WithName(worker3.GetName()),
-						server.WithAddress(worker3.GetAddress()),
-						server.WithOperationalState(server.ShutdownOperationalState.String()),
-						server.WithReleaseVersion("Boundary v.0.11"),
-						server.WithPublicId(worker3.GetPublicId())))
-			},
-			wantCnt:   0,
-			wantState: server.ShutdownOperationalState.String(),
-		},
-		{ // Upsert without a release version or state and expect to get a hit- test backwards compatibility
-			// Pre 0.11 workers will default to Active
-			name: "upsert-no-release-version-no-state",
-			upsertFn: func() (*server.Worker, error) {
-				return serversRepo.UpsertWorkerStatus(ctx,
-					server.NewWorker(scope.Global.String(),
-						server.WithName(worker3.GetName()),
-						server.WithAddress(worker3.GetAddress())),
-					server.WithPublicId(worker3.GetPublicId()))
-			},
-			wantCnt:   1,
-			wantState: server.ActiveOperationalState.String(),
-		},
-		{ // Upsert with active status and no version and expect to get a hit- test backwards compatibility
-			name: "upsert-no-release-version-active-state",
-			upsertFn: func() (*server.Worker, error) {
-				return serversRepo.UpsertWorkerStatus(ctx,
-					server.NewWorker(scope.Global.String(),
-						server.WithName(worker3.GetName()),
-						server.WithAddress(worker3.GetAddress()),
-						server.WithOperationalState(server.ActiveOperationalState.String())),
-					server.WithPublicId(worker3.GetPublicId()))
-			},
-			wantCnt:   1,
-			wantState: server.ActiveOperationalState.String(),
-		},
-		{ // Upsert with unknown status and do not expect to get a hit- test worker create before status
-			name: "upsert-unknown-status",
-			upsertFn: func() (*server.Worker, error) {
-				return serversRepo.UpsertWorkerStatus(ctx,
-					server.NewWorker(scope.Global.String(),
-						server.WithName(worker3.GetName()),
-						server.WithAddress(worker3.GetAddress()),
-						server.WithOperationalState(server.UnknownOperationalState.String())),
-					server.WithPublicId(worker3.GetPublicId()))
-			},
-			wantCnt:   0,
-			wantState: server.UnknownOperationalState.String(),
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			worker, err := tt.upsertFn()
-			require.NoError(err)
-			got, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithActiveWorkers(true))
-			require.NoError(err)
-			assert.Len(t, got, tt.wantCnt)
-			if len(tt.wantState) > 0 {
-				assert.Equal(t, tt.wantState, worker.OperationalState)
-			}
-		})
-	}
-}
-
-func TestListWorkers_WithLiveness(t *testing.T) {
-	t.Parallel()
-	ctx := context.Background()
-	require := require.New(t)
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	serversRepo, err := server.NewRepository(ctx, rw, rw, kms)
-	require.NoError(err)
-
-	worker1 := server.TestKmsWorker(t, conn, wrapper)
-	worker2 := server.TestKmsWorker(t, conn, wrapper)
-	worker3 := server.TestKmsWorker(t, conn, wrapper)
-
-	// Sleep the default liveness time (15sec currently) +1s
-	time.Sleep(time.Second * 16)
-
-	// Push an upsert to the first worker so that its status has been
-	// updated.
-	_, err = serversRepo.UpsertWorkerStatus(ctx,
-		server.NewWorker(scope.Global.String(),
-			server.WithName(worker1.GetName()),
-			server.WithAddress(worker1.GetAddress())),
-		server.WithPublicId(worker1.GetPublicId()))
-	require.NoError(err)
-
-	requireIds := func(expected []string, actual []*server.Worker) {
-		require.Len(expected, len(actual))
-		want := make(map[string]struct{})
-		for _, v := range expected {
-			want[v] = struct{}{}
-		}
-
-		got := make(map[string]struct{})
-		for _, v := range actual {
-			got[v.PublicId] = struct{}{}
-		}
-
-		require.Equal(want, got)
-	}
-
-	// Default liveness, should only list 1
-	result, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()})
-	require.NoError(err)
-	require.Len(result, 1)
-	requireIds([]string{worker1.GetPublicId()}, result)
-
-	// Upsert second server.
-	_, err = serversRepo.UpsertWorkerStatus(ctx,
-		server.NewWorker(scope.Global.String(),
-			server.WithName(worker2.GetName()),
-			server.WithAddress(worker2.GetAddress())),
-		server.WithPublicId(worker2.GetPublicId()))
-	require.NoError(err)
-
-	// Static liveness. Should get two, so long as this did not take
-	// more than 5s to execute.
-	result, err = serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithLiveness(time.Second*5))
-	require.NoError(err)
-	require.Len(result, 2)
-	requireIds([]string{worker1.GetPublicId(), worker2.GetPublicId()}, result)
-
-	// Liveness disabled, should get all three workers.
-	result, err = serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithLiveness(-1))
-	require.NoError(err)
-	require.Len(result, 3)
-	requireIds([]string{worker1.GetPublicId(), worker2.GetPublicId(), worker3.GetPublicId()}, result)
-}
-
 func TestRepository_CreateWorker(t *testing.T) {
 	t.Parallel()
 	testCtx := context.Background()
@@ -1668,3 +1371,389 @@ func TestRepository_UpdateWorker(t *testing.T) {
 		})
 	}
 }
+
+func TestListHcpbManagedWorkers(t *testing.T) {
+	t.Parallel()
+
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+
+	ctx := context.Background()
+	repo, err := server.NewRepository(ctx, rw, rw, kms.TestKms(t, conn, wrapper))
+	require.NoError(t, err)
+
+	hcpbTag := &server.Tag{Key: server.ManagedWorkerTag, Value: "true"}
+
+	t.Run("invalidLiveness", func(t *testing.T) {
+		worker := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(hcpbTag))
+		t.Cleanup(func() {
+			_, err = repo.DeleteWorker(ctx, worker.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		_, err := repo.UpsertWorkerStatus(ctx,
+			server.NewWorker(
+				scope.Global.String(),
+				server.WithName(worker.GetName()),
+				server.WithAddress(worker.GetAddress()),
+			),
+			server.WithPublicId(worker.GetPublicId()),
+		)
+		require.NoError(t, err)
+
+		workers, err := repo.ListHcpbManagedWorkers(ctx, -10)
+		require.NoError(t, err)
+		require.Len(t, workers, 1)
+		require.Equal(t, workers[0].PublicId, worker.GetPublicId())
+		require.Equal(t, workers[0].Address, worker.GetAddress())
+	})
+
+	t.Run("outsideLivenessThreshold", func(t *testing.T) {
+		worker := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(hcpbTag))
+		t.Cleanup(func() {
+			_, err = repo.DeleteWorker(ctx, worker.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		_, err := repo.UpsertWorkerStatus(ctx,
+			server.NewWorker(
+				scope.Global.String(),
+				server.WithName(worker.GetName()),
+				server.WithAddress(worker.GetAddress()),
+			),
+			server.WithPublicId(worker.GetPublicId()),
+		)
+		require.NoError(t, err)
+
+		<-time.After(2 * time.Second)
+		workers, err := repo.ListHcpbManagedWorkers(ctx, time.Second)
+		require.NoError(t, err)
+		require.Len(t, workers, 0)
+	})
+
+	t.Run("inLivenessThreshold", func(t *testing.T) {
+		worker := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(hcpbTag))
+		t.Cleanup(func() {
+			_, err = repo.DeleteWorker(ctx, worker.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		_, err := repo.UpsertWorkerStatus(ctx,
+			server.NewWorker(
+				scope.Global.String(),
+				server.WithName(worker.GetName()),
+				server.WithAddress(worker.GetAddress()),
+			),
+			server.WithPublicId(worker.GetPublicId()),
+		)
+		require.NoError(t, err)
+
+		workers, err := repo.ListHcpbManagedWorkers(ctx, time.Minute)
+		require.NoError(t, err)
+		require.Len(t, workers, 1)
+		require.Equal(t, workers[0].PublicId, worker.GetPublicId())
+		require.Equal(t, workers[0].Address, worker.GetAddress())
+	})
+
+	t.Run("multipleWorkers", func(t *testing.T) {
+		// Not HCPb managed but in liveness interval.
+		worker1 := server.TestKmsWorker(t, conn, wrapper)
+		t.Cleanup(func() {
+			_, err = repo.DeleteWorker(ctx, worker1.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		// HCPb managed not in liveness interval.
+		worker2 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(hcpbTag))
+		t.Cleanup(func() {
+			_, err = repo.DeleteWorker(ctx, worker2.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		// HCPb managed in liveness interval.
+		worker3 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(hcpbTag))
+		t.Cleanup(func() {
+			_, err = repo.DeleteWorker(ctx, worker3.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		// HCPb managed in liveness interval.
+		worker4 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(hcpbTag))
+		t.Cleanup(func() {
+			_, err = repo.DeleteWorker(ctx, worker4.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		<-time.After(2 * time.Second)
+
+		// Update all the workers that are meant to be within the liveness
+		// interval (worker1, worker3, worker4).
+		_, err := repo.UpsertWorkerStatus(ctx,
+			server.NewWorker(
+				scope.Global.String(),
+				server.WithName(worker1.GetName()),
+				server.WithAddress(worker1.GetAddress()),
+			),
+			server.WithPublicId(worker1.GetPublicId()),
+		)
+		require.NoError(t, err)
+
+		_, err = repo.UpsertWorkerStatus(ctx,
+			server.NewWorker(
+				scope.Global.String(),
+				server.WithName(worker3.GetName()),
+				server.WithAddress(worker3.GetAddress()),
+			),
+			server.WithPublicId(worker3.GetPublicId()),
+		)
+		require.NoError(t, err)
+
+		_, err = repo.UpsertWorkerStatus(ctx,
+			server.NewWorker(
+				scope.Global.String(),
+				server.WithName(worker4.GetName()),
+				server.WithAddress(worker4.GetAddress()),
+			),
+			server.WithPublicId(worker4.GetPublicId()),
+		)
+		require.NoError(t, err)
+
+		// List should return worker3 and worker4.
+		workers, err := repo.ListHcpbManagedWorkers(ctx, time.Second)
+		require.NoError(t, err)
+
+		exp := []server.WorkerAddress{
+			{PublicId: worker3.GetPublicId(), Address: worker3.GetAddress()},
+			{PublicId: worker4.GetPublicId(), Address: worker4.GetAddress()},
+		}
+		require.ElementsMatch(t, exp, workers)
+	})
+}
+
+func TestFilterWorkers_EgressFilter(t *testing.T) {
+	ctx := context.Background()
+	// This prevents us from running tests in parallel.
+	server.TestUseCommunityFilterWorkersFn(t)
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	kmsCache := kms.TestKms(t, conn, wrapper)
+	require.NoError(t, kmsCache.CreateKeys(context.Background(), scope.Global.String(), kms.WithRandomReader(rand.Reader)))
+
+	var workers []*server.Worker
+	for i := 0; i < 5; i++ {
+		switch {
+		case i%2 == 0:
+			workers = append(workers, server.TestKmsWorker(t, conn, wrapper,
+				server.WithName(fmt.Sprintf("test_worker_%d", i)),
+				server.WithWorkerTags(&server.Tag{
+					Key:   fmt.Sprintf("key%d", i),
+					Value: fmt.Sprintf("value%d", i),
+				})))
+		default:
+			workers = append(workers, server.TestPkiWorker(t, conn, wrapper,
+				server.WithName(fmt.Sprintf("test_worker_%d", i)),
+				server.WithWorkerTags(&server.Tag{
+					Key:   "key",
+					Value: "configvalue",
+				})))
+		}
+	}
+
+	cases := []struct {
+		name        string
+		in          []*server.Worker
+		out         []*server.Worker
+		filter      string
+		errContains string
+	}{
+		{
+			name:        "no-workers",
+			in:          []*server.Worker{},
+			out:         []*server.Worker{},
+			filter:      "",
+			errContains: "No workers are available to handle this session, or all have been filtered",
+		},
+		{
+			name: "no-filter",
+			in:   workers,
+			out:  workers,
+		},
+		{
+			name:        "filter-no-matches",
+			in:          workers,
+			out:         workers,
+			filter:      `"/name" matches "test_worker_[13]" and "configvalue2" in "/tags/key"`,
+			errContains: "No workers are available to handle this session, or all have been filtered",
+		},
+		{
+			name:   "filter-one-match",
+			in:     workers,
+			out:    []*server.Worker{workers[1]},
+			filter: `"/name" matches "test_worker_[12]" and "configvalue" in "/tags/key"`,
+		},
+		{
+			name:   "filter-two-matches",
+			in:     workers,
+			out:    []*server.Worker{workers[1], workers[3]},
+			filter: `"configvalue" in "/tags/key"`,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			target := &targettest.Target{
+				Target: &tgstore.Target{},
+			}
+			if len(tc.filter) > 0 {
+				target.EgressWorkerFilter = tc.filter
+			}
+
+			out, protocolWorker, err := server.FilterWorkersFn(
+				ctx, nil, target, tc.in, "", nil, nil, nil,
+			)
+			if tc.errContains != "" {
+				require.ErrorContains(err, tc.errContains)
+				assert.Nil(out)
+				return
+			}
+
+			require.NoError(err)
+			require.Len(out, len(tc.out))
+			for i, exp := range tc.out {
+				assert.Equal(exp.Name, out[i].Name)
+			}
+			require.Nil(protocolWorker)
+		})
+	}
+}
+
+func TestSelectSessionWorkers(t *testing.T) {
+	// This prevents us from running tests in parallel.
+	server.TestUseCommunityFilterWorkersFn(t)
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+
+	wrapper := db.TestWrapper(t)
+	kmsCache := kms.TestKms(t, conn, wrapper)
+	err := kmsCache.CreateKeys(context.Background(), scope.Global.String(), kms.WithRandomReader(rand.Reader))
+	require.NoError(t, err)
+
+	ctx := context.Background()
+
+	iamRepo, err := iam.NewRepository(ctx, rw, rw, kmsCache)
+	require.NoError(t, err)
+	_, proj := iam.TestScopes(t, iamRepo)
+	require.NotNil(t, proj)
+
+	repo, err := server.NewRepository(ctx, rw, rw, kmsCache)
+	require.NoError(t, err)
+	require.NotNil(t, repo)
+
+	t.Run("noAvailableWorkers", func(t *testing.T) {
+		tg, err := target.New(ctx, targettest.Subtype, proj.GetPublicId(), target.WithWorkerFilter(`"doesnt_exist" in "/tags/type"`))
+		require.NoError(t, err)
+		require.NotNil(t, tg)
+
+		was, protoWorkerId, err := repo.SelectSessionWorkers(ctx, server.DefaultLiveness, tg, "", nil, nil, nil)
+		require.Nil(t, was)
+		require.Empty(t, protoWorkerId)
+		require.ErrorContains(t, err, "No workers are available to handle this session.")
+	})
+
+	t.Run("invalidWorkerStatusGracePeriod", func(t *testing.T) {
+		w1 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: "type", Value: "worker1"}))
+		require.NotNil(t, w1)
+		t.Cleanup(func() {
+			_, err := repo.DeleteWorker(ctx, w1.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		w2 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: "type", Value: "worker2"}))
+		require.NotNil(t, w2)
+		t.Cleanup(func() {
+			_, err := repo.DeleteWorker(ctx, w2.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		tg, err := target.New(ctx, targettest.Subtype, proj.GetPublicId(), target.WithWorkerFilter(`"worker1" in "/tags/type"`))
+		require.NoError(t, err)
+		require.NotNil(t, tg)
+
+		was, protoWorkerId, err := repo.SelectSessionWorkers(ctx, -10, tg, "", nil, nil, nil)
+		require.NoError(t, err)
+		require.Empty(t, protoWorkerId)
+		require.Len(t, was, 1)
+		require.Equal(t, w1.GetPublicId(), was[0].PublicId)
+		require.Equal(t, w1.GetAddress(), was[0].Address)
+	})
+
+	t.Run("validWorkerStatusGracePeriod", func(t *testing.T) {
+		w1 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: "type", Value: "prod"}))
+		require.NotNil(t, w1)
+		t.Cleanup(func() {
+			_, err := repo.DeleteWorker(ctx, w1.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		w2 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: "type", Value: "prod"}))
+		require.NotNil(t, w2)
+		t.Cleanup(func() {
+			_, err := repo.DeleteWorker(ctx, w2.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		<-time.After(2 * time.Second)
+		_, err := repo.UpsertWorkerStatus(ctx,
+			server.NewWorker(
+				scope.Global.String(),
+				server.WithName(w1.GetName()),
+				server.WithAddress(w1.GetAddress()),
+			),
+			server.WithPublicId(w1.GetPublicId()),
+		)
+		require.NoError(t, err)
+
+		tg, err := target.New(ctx, targettest.Subtype, proj.GetPublicId(), target.WithWorkerFilter(`"prod" in "/tags/type"`))
+		require.NoError(t, err)
+		require.NotNil(t, tg)
+
+		was, protoWorkerId, err := repo.SelectSessionWorkers(ctx, time.Second, tg, "", nil, nil, nil)
+		require.NoError(t, err)
+		require.Empty(t, protoWorkerId)
+		require.Len(t, was, 1)
+		require.Equal(t, w1.GetPublicId(), was[0].PublicId)
+		require.Equal(t, w1.GetAddress(), was[0].Address)
+	})
+
+	t.Run("multipleWorkers", func(t *testing.T) {
+		w1 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: "type", Value: "prod"}))
+		require.NotNil(t, w1)
+		t.Cleanup(func() {
+			_, err := repo.DeleteWorker(ctx, w1.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		w2 := server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: "type", Value: "prod"}))
+		require.NotNil(t, w2)
+		t.Cleanup(func() {
+			_, err := repo.DeleteWorker(ctx, w2.GetPublicId())
+			assert.NoError(t, err)
+		})
+
+		tg, err := target.New(ctx, targettest.Subtype, proj.GetPublicId(), target.WithWorkerFilter(`"prod" in "/tags/type"`))
+		require.NoError(t, err)
+		require.NotNil(t, tg)
+
+		was, protoWorkerId, err := repo.SelectSessionWorkers(ctx, server.DefaultLiveness, tg, "", nil, nil, nil)
+		require.NoError(t, err)
+		require.Empty(t, protoWorkerId)
+		require.Len(t, was, 2)
+
+		exp := []server.WorkerAddress{
+			{PublicId: w1.GetPublicId(), Address: w1.GetAddress()},
+			{PublicId: w2.GetPublicId(), Address: w2.GetAddress()},
+		}
+		require.ElementsMatch(t, exp, was)
+	})
+}

internal/server/store/gorm.go

@@ -8,7 +8,12 @@ func (*Controller) TableName() string {
 	return "server_controller"
 }
 
-// TableName overrides the table name used by WorkerTag to `worker_tag`
-func (*WorkerTag) TableName() string {
-	return "server_worker_tag"
+// TableName overrides the table name used by ApiTag
+func (w *ApiTag) TableName() string {
+	return "server_worker_api_tag"
+}
+
+// TableName overrides the table name used by ConfigTag
+func (w *ConfigTag) TableName() string {
+	return "server_worker_config_tag"
 }

internal/server/store/worker.pb.go

@@ -204,6 +204,9 @@ func (x *Worker) GetLocalStorageState() string {
 
 // WorkerTag is a tag for a worker.  The primary key is comprised of the
 // worker_id, key, value, and source.
+// WorkerTag is deprecated- use ApiTag and ConfigTag instead.
+//
+// Deprecated: Marked as deprecated in controller/storage/servers/store/v1/worker.proto.
 type WorkerTag struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -283,6 +286,148 @@ func (x *WorkerTag) GetSource() string {
 	return ""
 }
 
+// ApiTag is an API tag for a worker. The primary key is comprised of the
+// worker_id, key, and value
+type ApiTag struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// worker_id is the public key that key of the worker this tag is for.
+	// @inject_tag: `gorm:"primary_key"`
+	WorkerId string `protobuf:"bytes,10,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty" gorm:"primary_key"`
+	// key is the key of the tag. This must be set.
+	// @inject_tag: `gorm:"primary_key"`
+	Key string `protobuf:"bytes,20,opt,name=key,proto3" json:"key,omitempty" gorm:"primary_key"`
+	// value is the value
+	// @inject_tag: `gorm:"primary_key"`
+	Value string `protobuf:"bytes,30,opt,name=value,proto3" json:"value,omitempty" gorm:"primary_key"`
+}
+
+func (x *ApiTag) Reset() {
+	*x = ApiTag{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ApiTag) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ApiTag) ProtoMessage() {}
+
+func (x *ApiTag) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ApiTag.ProtoReflect.Descriptor instead.
+func (*ApiTag) Descriptor() ([]byte, []int) {
+	return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *ApiTag) GetWorkerId() string {
+	if x != nil {
+		return x.WorkerId
+	}
+	return ""
+}
+
+func (x *ApiTag) GetKey() string {
+	if x != nil {
+		return x.Key
+	}
+	return ""
+}
+
+func (x *ApiTag) GetValue() string {
+	if x != nil {
+		return x.Value
+	}
+	return ""
+}
+
+// ConfigTag is a configuration tag for a worker. The primary key is comprised of the
+// worker_id, key, and value
+type ConfigTag struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// worker_id is the public key that key of the worker this tag is for.
+	// @inject_tag: `gorm:"primary_key"`
+	WorkerId string `protobuf:"bytes,10,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty" gorm:"primary_key"`
+	// key is the key of the tag. This must be set.
+	// @inject_tag: `gorm:"primary_key"`
+	Key string `protobuf:"bytes,20,opt,name=key,proto3" json:"key,omitempty" gorm:"primary_key"`
+	// value is the value
+	// @inject_tag: `gorm:"primary_key"`
+	Value string `protobuf:"bytes,30,opt,name=value,proto3" json:"value,omitempty" gorm:"primary_key"`
+}
+
+func (x *ConfigTag) Reset() {
+	*x = ConfigTag{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ConfigTag) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ConfigTag) ProtoMessage() {}
+
+func (x *ConfigTag) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ConfigTag.ProtoReflect.Descriptor instead.
+func (*ConfigTag) Descriptor() ([]byte, []int) {
+	return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *ConfigTag) GetWorkerId() string {
+	if x != nil {
+		return x.WorkerId
+	}
+	return ""
+}
+
+func (x *ConfigTag) GetKey() string {
+	if x != nil {
+		return x.Key
+	}
+	return ""
+}
+
+func (x *ConfigTag) GetValue() string {
+	if x != nil {
+		return x.Value
+	}
+	return ""
+}
+
 // WorkerStorageBucketCredentialState is a state for a storage bucket credential for a worker.
 // The primary key is comprised of the worker_id, storage_bucket_credential_id, permission_type.
 type WorkerStorageBucketCredentialState struct {
@@ -312,7 +457,7 @@ type WorkerStorageBucketCredentialState struct {
 func (x *WorkerStorageBucketCredentialState) Reset() {
 	*x = WorkerStorageBucketCredentialState{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[2]
+		mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[4]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -325,7 +470,7 @@ func (x *WorkerStorageBucketCredentialState) String() string {
 func (*WorkerStorageBucketCredentialState) ProtoMessage() {}
 
 func (x *WorkerStorageBucketCredentialState) ProtoReflect() protoreflect.Message {
-	mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[2]
+	mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[4]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -338,7 +483,7 @@ func (x *WorkerStorageBucketCredentialState) ProtoReflect() protoreflect.Message
 
 // Deprecated: Use WorkerStorageBucketCredentialState.ProtoReflect.Descriptor instead.
 func (*WorkerStorageBucketCredentialState) Descriptor() ([]byte, []int) {
-	return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{2}
+	return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{4}
 }
 
 func (x *WorkerStorageBucketCredentialState) GetWorkerId() string {
@@ -436,37 +581,47 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{
 	0x12, 0x2f, 0x0a, 0x13, 0x6c, 0x6f, 0x63, 0x61, 0x6c, 0x5f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67,
 	0x65, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x65, 0x18, 0xa0, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x11,
 	0x6c, 0x6f, 0x63, 0x61, 0x6c, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x53, 0x74, 0x61, 0x74,
-	0x65, 0x22, 0x68, 0x0a, 0x09, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x54, 0x61, 0x67, 0x12, 0x1b,
+	0x65, 0x22, 0x6c, 0x0a, 0x09, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x54, 0x61, 0x67, 0x12, 0x1b,
 	0x0a, 0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28,
 	0x09, 0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x6b,
 	0x65, 0x79, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a,
 	0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61,
 	0x6c, 0x75, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x28, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0xb1, 0x02, 0x0a, 0x22,
-	0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63,
-	0x6b, 0x65, 0x74, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x53, 0x74, 0x61,
-	0x74, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18,
-	0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12,
-	0x3f, 0x0a, 0x1c, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x5f, 0x62, 0x75, 0x63, 0x6b, 0x65,
-	0x74, 0x5f, 0x63, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x5f, 0x69, 0x64, 0x18,
-	0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x19, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75,
-	0x63, 0x6b, 0x65, 0x74, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x49, 0x64,
-	0x12, 0x27, 0x0a, 0x0f, 0x70, 0x65, 0x72, 0x6d, 0x69, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x5f, 0x74,
-	0x79, 0x70, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x70, 0x65, 0x72, 0x6d, 0x69,
-	0x73, 0x73, 0x69, 0x6f, 0x6e, 0x54, 0x79, 0x70, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x73, 0x74, 0x61,
-	0x74, 0x65, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x73, 0x74, 0x61, 0x74, 0x65, 0x12,
-	0x23, 0x0a, 0x0d, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x5f, 0x64, 0x65, 0x74, 0x61, 0x69, 0x6c, 0x73,
-	0x18, 0x32, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x44, 0x65, 0x74,
-	0x61, 0x69, 0x6c, 0x73, 0x12, 0x49, 0x0a, 0x0a, 0x63, 0x68, 0x65, 0x63, 0x6b, 0x65, 0x64, 0x5f,
-	0x61, 0x74, 0x18, 0x3c, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72,
-	0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69,
-	0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73,
-	0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x63, 0x68, 0x65, 0x63, 0x6b, 0x65, 0x64, 0x41, 0x74, 0x42,
-	0x3b, 0x5a, 0x39, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61,
-	0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79,
-	0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
-	0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x33,
+	0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x3a, 0x02, 0x18, 0x01, 0x22,
+	0x4d, 0x0a, 0x06, 0x41, 0x70, 0x69, 0x54, 0x61, 0x67, 0x12, 0x1b, 0x0a, 0x09, 0x77, 0x6f, 0x72,
+	0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x77, 0x6f,
+	0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x14, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75,
+	0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x50,
+	0x0a, 0x09, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x54, 0x61, 0x67, 0x12, 0x1b, 0x0a, 0x09, 0x77,
+	0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08,
+	0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18,
+	0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61,
+	0x6c, 0x75, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65,
+	0x22, 0xb1, 0x02, 0x0a, 0x22, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x6f, 0x72, 0x61,
+	0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69,
+	0x61, 0x6c, 0x53, 0x74, 0x61, 0x74, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65,
+	0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b,
+	0x65, 0x72, 0x49, 0x64, 0x12, 0x3f, 0x0a, 0x1c, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x5f,
+	0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x5f, 0x63, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61,
+	0x6c, 0x5f, 0x69, 0x64, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x19, 0x73, 0x74, 0x6f, 0x72,
+	0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74,
+	0x69, 0x61, 0x6c, 0x49, 0x64, 0x12, 0x27, 0x0a, 0x0f, 0x70, 0x65, 0x72, 0x6d, 0x69, 0x73, 0x73,
+	0x69, 0x6f, 0x6e, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e,
+	0x70, 0x65, 0x72, 0x6d, 0x69, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x54, 0x79, 0x70, 0x65, 0x12, 0x14,
+	0x0a, 0x05, 0x73, 0x74, 0x61, 0x74, 0x65, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x73,
+	0x74, 0x61, 0x74, 0x65, 0x12, 0x23, 0x0a, 0x0d, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x5f, 0x64, 0x65,
+	0x74, 0x61, 0x69, 0x6c, 0x73, 0x18, 0x32, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x65, 0x72, 0x72,
+	0x6f, 0x72, 0x44, 0x65, 0x74, 0x61, 0x69, 0x6c, 0x73, 0x12, 0x49, 0x0a, 0x0a, 0x63, 0x68, 0x65,
+	0x63, 0x6b, 0x65, 0x64, 0x5f, 0x61, 0x74, 0x18, 0x3c, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e,
+	0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61,
+	0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e,
+	0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x63, 0x68, 0x65, 0x63, 0x6b,
+	0x65, 0x64, 0x41, 0x74, 0x42, 0x3b, 0x5a, 0x39, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
+	0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75,
+	0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73,
+	0x65, 0x72, 0x76, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72,
+	0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -481,18 +636,20 @@ func file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP() []byte
 	return file_controller_storage_servers_store_v1_worker_proto_rawDescData
 }
 
-var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
 var file_controller_storage_servers_store_v1_worker_proto_goTypes = []any{
 	(*Worker)(nil),    // 0: controller.storage.servers.store.v1.Worker
 	(*WorkerTag)(nil), // 1: controller.storage.servers.store.v1.WorkerTag
-	(*WorkerStorageBucketCredentialState)(nil), // 2: controller.storage.servers.store.v1.WorkerStorageBucketCredentialState
-	(*timestamp.Timestamp)(nil),                // 3: controller.storage.timestamp.v1.Timestamp
+	(*ApiTag)(nil),    // 2: controller.storage.servers.store.v1.ApiTag
+	(*ConfigTag)(nil), // 3: controller.storage.servers.store.v1.ConfigTag
+	(*WorkerStorageBucketCredentialState)(nil), // 4: controller.storage.servers.store.v1.WorkerStorageBucketCredentialState
+	(*timestamp.Timestamp)(nil),                // 5: controller.storage.timestamp.v1.Timestamp
 }
 var file_controller_storage_servers_store_v1_worker_proto_depIdxs = []int32{
-	3, // 0: controller.storage.servers.store.v1.Worker.create_time:type_name -> controller.storage.timestamp.v1.Timestamp
-	3, // 1: controller.storage.servers.store.v1.Worker.update_time:type_name -> controller.storage.timestamp.v1.Timestamp
-	3, // 2: controller.storage.servers.store.v1.Worker.last_status_time:type_name -> controller.storage.timestamp.v1.Timestamp
-	3, // 3: controller.storage.servers.store.v1.WorkerStorageBucketCredentialState.checked_at:type_name -> controller.storage.timestamp.v1.Timestamp
+	5, // 0: controller.storage.servers.store.v1.Worker.create_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	5, // 1: controller.storage.servers.store.v1.Worker.update_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	5, // 2: controller.storage.servers.store.v1.Worker.last_status_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	5, // 3: controller.storage.servers.store.v1.WorkerStorageBucketCredentialState.checked_at:type_name -> controller.storage.timestamp.v1.Timestamp
 	4, // [4:4] is the sub-list for method output_type
 	4, // [4:4] is the sub-list for method input_type
 	4, // [4:4] is the sub-list for extension type_name
@@ -531,6 +688,30 @@ func file_controller_storage_servers_store_v1_worker_proto_init() {
 			}
 		}
 		file_controller_storage_servers_store_v1_worker_proto_msgTypes[2].Exporter = func(v any, i int) any {
+			switch v := v.(*ApiTag); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_controller_storage_servers_store_v1_worker_proto_msgTypes[3].Exporter = func(v any, i int) any {
+			switch v := v.(*ConfigTag); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_controller_storage_servers_store_v1_worker_proto_msgTypes[4].Exporter = func(v any, i int) any {
 			switch v := v.(*WorkerStorageBucketCredentialState); i {
 			case 0:
 				return &v.state
@@ -549,7 +730,7 @@ func file_controller_storage_servers_store_v1_worker_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_controller_storage_servers_store_v1_worker_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   3,
+			NumMessages:   5,
 			NumExtensions: 0,
 			NumServices:   0,
 		},

internal/server/testing.go

@@ -125,19 +125,19 @@ func TestKmsWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...O
 	}
 
 	wrk := NewWorker(scope.Global.String(), opt...)
-	wrk, err = serversRepo.UpsertWorkerStatus(ctx, wrk)
+	wrk, err = TestUpsertAndReturnWorker(ctx, t, wrk, serversRepo)
+	require.NoError(t, err)
 	require.NoError(t, err)
 	require.NotNil(t, wrk)
 	require.Equal(t, "kms", wrk.Type)
 
 	if len(opts.withWorkerTags) > 0 {
-		var tags []*store.WorkerTag
+		var tags []*store.ConfigTag
 		for _, t := range opts.withWorkerTags {
-			tags = append(tags, &store.WorkerTag{
+			tags = append(tags, &store.ConfigTag{
 				WorkerId: wrk.GetPublicId(),
 				Key:      t.Key,
 				Value:    t.Value,
-				Source:   "configuration",
 			})
 		}
 		require.NoError(t, rw.CreateItems(ctx, tags))
@@ -170,13 +170,12 @@ func TestPkiWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...O
 	require.NotNil(t, wrk)
 
 	if len(opts.withWorkerTags) > 0 {
-		var tags []*store.WorkerTag
+		var tags []*store.ConfigTag
 		for _, t := range opts.withWorkerTags {
-			tags = append(tags, &store.WorkerTag{
+			tags = append(tags, &store.ConfigTag{
 				WorkerId: wrk.GetPublicId(),
 				Key:      t.Key,
 				Value:    t.Value,
-				Source:   "configuration",
 			})
 		}
 		require.NoError(t, rw.CreateItems(ctx, tags))
@@ -212,3 +211,35 @@ func TestPkiWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...O
 	require.NoError(t, err)
 	return wrk
 }
+
+// TestLookupWorkerByName looks up a worker by name
+func TestLookupWorkerByName(ctx context.Context, t *testing.T, name string, serversRepo *Repository) (*Worker, error) {
+	workers, err := serversRepo.ListWorkers(ctx, []string{"global"})
+	require.NoError(t, err)
+	for _, w := range workers {
+		if w.GetName() == name {
+			return w, nil
+		}
+	}
+	return nil, nil
+}
+
+// TestUpsertAndReturnWorker upserts and returns a worker
+func TestUpsertAndReturnWorker(ctx context.Context, t *testing.T, w *Worker, serversRepo *Repository, opt ...Option) (*Worker, error) {
+	workerId, err := serversRepo.UpsertWorkerStatus(ctx, w, opt...)
+	require.NoError(t, err)
+	require.NotEmpty(t, workerId)
+	return serversRepo.LookupWorker(ctx, workerId)
+}
+
+// TestUseCommunityFilterWorkersFn is used to ensure that CE tests run from the
+// ENT repo use the CE worker filtering logic. WARNING: Do NOT run tests in
+// parallel when using this.
+func TestUseCommunityFilterWorkersFn(t *testing.T) {
+	oldFn := FilterWorkersFn
+	FilterWorkersFn = filterWorkers
+
+	t.Cleanup(func() {
+		FilterWorkersFn = oldFn
+	})
+}

internal/server/testing_test.go

@@ -9,12 +9,14 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/types/scope"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/testing/protocmp"
 )
 
 func TestTestKmsWorker(t *testing.T) {
@@ -71,3 +73,25 @@ func TestTestPkiWorker(t *testing.T) {
 	assert.True(t, strings.HasPrefix(authorizedWorker.GetPublicId(), globals.WorkerPrefix))
 	assert.NotEmpty(t, keyId)
 }
+
+func TestTestLookupWorkerByName(t *testing.T) {
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	repo, err := NewRepository(ctx, rw, rw, kms)
+	require.NoError(t, err)
+
+	w := TestKmsWorker(t, conn, wrapper)
+	t.Run("success", func(t *testing.T) {
+		got, err := TestLookupWorkerByName(ctx, t, w.GetName(), repo)
+		require.NoError(t, err)
+		assert.Empty(t, cmp.Diff(w.Worker, got.Worker, protocmp.Transform()))
+	})
+	t.Run("not found", func(t *testing.T) {
+		got, err := TestLookupWorkerByName(ctx, t, "unknown_name", repo)
+		require.NoError(t, err)
+		assert.Nil(t, got)
+	})
+}

internal/server/worker.go

@@ -5,7 +5,6 @@ package server
 
 import (
 	"context"
-	"strings"
 
 	"github.com/fatih/structs"
 	"github.com/hashicorp/boundary/internal/db/timestamp"
@@ -113,10 +112,9 @@ func AttachWorkerIdToState(ctx context.Context, workerId string) (*structpb.Stru
 // authorizing and establishing a session.  It is owned by a scope.
 type Worker struct {
 	*store.Worker
-
-	activeConnectionCount uint32 `gorm:"-"`
-	apiTags               []*Tag `gorm:"-"`
-	configTags            []*Tag `gorm:"-"`
+	ApiTags               Tags   `json:"api_tags" gorm:"->"`
+	ConfigTags            Tags   `json:"config_tags" gorm:"->"`
+	ActiveConnectionCount uint32 `gorm:"->"`
 
 	// inputTags is not specified to be api or config tags and is not intended
 	// to be read by clients.  Since config tags and api tags are applied in
@@ -150,7 +148,7 @@ func NewWorker(scopeId string, opt ...Option) *Worker {
 		inputTags: opts.withWorkerTags,
 	}
 	if opts.withTestUseInputTagsAsApiTags {
-		worker.apiTags = worker.inputTags
+		worker.ApiTags = worker.inputTags
 	}
 	return worker
 }
@@ -168,17 +166,11 @@ func (w *Worker) clone() *Worker {
 	cWorker := &Worker{
 		Worker: cw.(*store.Worker),
 	}
-	if w.apiTags != nil {
-		cWorker.apiTags = make([]*Tag, 0, len(w.apiTags))
-		for _, t := range w.apiTags {
-			cWorker.apiTags = append(cWorker.apiTags, &Tag{Key: t.Key, Value: t.Value})
-		}
+	if w.ApiTags != nil {
+		cWorker.ApiTags = w.ApiTags.clone()
 	}
-	if w.configTags != nil {
-		cWorker.configTags = make([]*Tag, 0, len(w.configTags))
-		for _, t := range w.configTags {
-			cWorker.configTags = append(cWorker.configTags, &Tag{Key: t.Key, Value: t.Value})
-		}
+	if w.ConfigTags != nil {
+		cWorker.ConfigTags = w.ConfigTags.clone()
 	}
 	if w.inputTags != nil {
 		cWorker.inputTags = make([]*Tag, 0, len(w.inputTags))
@@ -189,35 +181,18 @@ func (w *Worker) clone() *Worker {
 	return cWorker
 }
 
-// ActiveConnectionCount is the current number of sessions this worker is handling
-// according to the controllers.
-func (w *Worker) ActiveConnectionCount() uint32 {
-	return w.activeConnectionCount
-}
-
 // CanonicalTags is the deduplicated set of tags contained on both the resource
 // set over the API as well as the tags reported by the worker itself. This
 // function is guaranteed to return a non-nil map.
 func (w *Worker) CanonicalTags(opt ...Option) map[string][]string {
-	dedupedTags := make(map[Tag]struct{})
-	for _, t := range w.apiTags {
-		dedupedTags[*t] = struct{}{}
-	}
-	for _, t := range w.configTags {
-		dedupedTags[*t] = struct{}{}
-	}
-	tags := make(map[string][]string)
-	for t := range dedupedTags {
-		tags[t.Key] = append(tags[t.Key], t.Value)
-	}
-	return tags
+	return compactTags(&w.ApiTags, &w.ConfigTags)
 }
 
 // GetConfigTags returns the tags for this worker which has been set through
 // the worker daemon's configuration file.
 func (w *Worker) GetConfigTags() map[string][]string {
 	tags := make(map[string][]string)
-	for _, t := range w.configTags {
+	for _, t := range w.ConfigTags {
 		tags[t.Key] = append(tags[t.Key], t.Value)
 	}
 	return tags
@@ -226,7 +201,7 @@ func (w *Worker) GetConfigTags() map[string][]string {
 // GetApiTags returns the api tags which have been set for this worker.
 func (w *Worker) GetApiTags() map[string][]string {
 	tags := make(map[string][]string)
-	for _, t := range w.apiTags {
+	for _, t := range w.ApiTags {
 		tags[t.Key] = append(tags[t.Key], t.Value)
 	}
 	return tags
@@ -246,95 +221,3 @@ func (w *Worker) GetLastStatusTime() *timestamp.Timestamp {
 func (Worker) TableName() string {
 	return "server_worker"
 }
-
-// workerAggregate contains an aggregated view of the values associated with
-// a single worker.
-type workerAggregate struct {
-	PublicId              string `gorm:"primary_key"`
-	ScopeId               string
-	Name                  string
-	Description           string
-	CreateTime            *timestamp.Timestamp
-	UpdateTime            *timestamp.Timestamp
-	Address               string
-	Version               uint32
-	Type                  string
-	ReleaseVersion        string
-	ApiTags               string
-	ActiveConnectionCount uint32
-	OperationalState      string
-	LocalStorageState     string
-	// Config Fields
-	LastStatusTime   *timestamp.Timestamp
-	WorkerConfigTags string
-}
-
-func (a *workerAggregate) toWorker(ctx context.Context) (*Worker, error) {
-	const op = "server.(workerAggregate).toWorker"
-	worker := &Worker{
-		Worker: &store.Worker{
-			PublicId:          a.PublicId,
-			Name:              a.Name,
-			Description:       a.Description,
-			Address:           a.Address,
-			CreateTime:        a.CreateTime,
-			UpdateTime:        a.UpdateTime,
-			ScopeId:           a.ScopeId,
-			Version:           a.Version,
-			LastStatusTime:    a.LastStatusTime,
-			Type:              a.Type,
-			ReleaseVersion:    a.ReleaseVersion,
-			OperationalState:  a.OperationalState,
-			LocalStorageState: a.LocalStorageState,
-		},
-		activeConnectionCount: a.ActiveConnectionCount,
-		RemoteStorageStates:   map[string]*plugin.StorageBucketCredentialState{},
-	}
-	tags, err := tagsFromAggregatedTagString(ctx, a.ApiTags)
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error parsing config tag string"))
-	}
-	worker.apiTags = tags
-
-	tags, err = tagsFromAggregatedTagString(ctx, a.WorkerConfigTags)
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error parsing config tag string"))
-	}
-	worker.configTags = tags
-
-	return worker, nil
-}
-
-// tagsForAggregatedTagString parses a deliminated string in the format returned
-// by the database for the server_worker_aggregate view and returns []*Tag.
-// The string is in the format of key1Yvalue1Zkey2Yvalue2Zkey3Yvalue3. Y and Z
-// ares chosen for deliminators since tag keys and values are restricted from
-// having capitalized letters in them.
-func tagsFromAggregatedTagString(ctx context.Context, s string) ([]*Tag, error) {
-	if s == "" {
-		return nil, nil
-	}
-	const op = "server.tagsFromAggregatedTagString"
-	const aggregateDelimiter = "Z"
-	const pairDelimiter = "Y"
-	var tags []*Tag
-	for _, kv := range strings.Split(s, aggregateDelimiter) {
-		res := strings.SplitN(kv, pairDelimiter, 3)
-		if len(res) != 2 {
-			return nil, errors.New(ctx, errors.Internal, op, "invalid aggregated tag pairs")
-		}
-		tags = append(tags, &Tag{
-			Key:   res[0],
-			Value: res[1],
-		})
-	}
-	return tags, nil
-}
-
-func (a *workerAggregate) GetPublicId() string {
-	return a.PublicId
-}
-
-func (workerAggregate) TableName() string {
-	return "server_worker_aggregate"
-}

internal/server/worker_list.go

@@ -4,14 +4,20 @@
 package server
 
 import (
+	"context"
+	"crypto/rand"
 	stderrors "errors"
 	"fmt"
+	"math/big"
+	"strings"
 
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
+	"github.com/hashicorp/boundary/internal/event"
 	pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets"
 	"github.com/hashicorp/boundary/sdk/pbs/plugin"
 	"github.com/hashicorp/boundary/version"
 	"github.com/hashicorp/go-bexpr"
+	gvers "github.com/hashicorp/go-version"
 	"github.com/mitchellh/pointerstructure"
 	"google.golang.org/grpc/codes"
 )
@@ -62,6 +68,31 @@ func (w WorkerList) SupportsFeature(f version.Feature) WorkerList {
 	return ret
 }
 
+// Shuffle returns a randomly-shuffled copy of the caller's Workers (using
+// crypto/rand). If the caller's WorkerList has one element or less, this
+// function is a no-op.
+func (w WorkerList) Shuffle() (WorkerList, error) {
+	if len(w) <= 1 {
+		return w, nil
+	}
+
+	ret := make(WorkerList, len(w))
+	copy(ret, w)
+
+	// This is an adaptation of the Fisher-Yates shuffle used in
+	// math/rand.Shuffle, but using the crypto/rand package instead. The same
+	// caveats as math/rand.Shuffle apply.
+	for i := len(ret) - 1; i > 0; i-- {
+		j, err := rand.Int(rand.Reader, big.NewInt(int64(i+1)))
+		if err != nil {
+			return nil, err
+		}
+		ret[i], ret[j.Uint64()] = ret[j.Uint64()], ret[i]
+	}
+
+	return ret, nil
+}
+
 // filtered returns a new workerList where all elements contained in it are the
 // ones which from the original workerList that pass the evaluator's evaluation.
 func (w WorkerList) Filtered(eval *bexpr.Evaluator) (WorkerList, error) {
@@ -84,6 +115,51 @@ func (w WorkerList) Filtered(eval *bexpr.Evaluator) (WorkerList, error) {
 	return ret, nil
 }
 
+// FilteredWithFeatures returns a new workerList where all elements contained in
+// it are the ones which from the original workerList that pass the evaluator's
+// evaluation and satisfy the features required.
+func (w WorkerList) FilteredWithFeatures(ctx context.Context, eval *bexpr.Evaluator, features []version.Feature) (WorkerList, error) {
+	const op = "server.WorkerList.FilteredWithFeatures"
+	var ret []*Worker
+workerLoop:
+	for _, worker := range w {
+		filterInput := map[string]interface{}{
+			"name": worker.GetName(),
+			"tags": worker.CanonicalTags(),
+		}
+		ok, err := eval.Evaluate(filterInput)
+		if err != nil && !stderrors.Is(err, pointerstructure.ErrNotFound) {
+			// If we find pointerstructure.ErrNotFound, don't error out but
+			// ignore the worker and go to the next.
+			return nil, err
+		}
+		if !ok {
+			continue
+		}
+		if len(features) > 0 {
+			versionString := worker.ReleaseVersion
+			idx := strings.Index(versionString, version.BoundaryPrefix)
+			if idx >= 0 {
+				versionString = versionString[idx+len(version.BoundaryPrefix):]
+			}
+			nodeVersion, err := gvers.NewVersion(versionString)
+			if err != nil {
+				// Emit error and continue, as we might still find a different worker
+				event.WriteError(ctx, op, fmt.Errorf("cannot parse worker version %s for worker %s", versionString, worker.Name))
+				continue
+			}
+			for _, f := range features {
+				if !version.SupportsFeature(nodeVersion, f) {
+					continue workerLoop
+				}
+			}
+		}
+
+		ret = append(ret, worker)
+	}
+	return ret, nil
+}
+
 // SeparateManagedWorkers divides the incoming workers into managed and
 // unmanaged workers, respectively
 func SeparateManagedWorkers(workers WorkerList) (managedWorkers, nonManagedWorkers WorkerList) {

internal/server/worker_list_test.go

@@ -0,0 +1,338 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package server
+
+import (
+	"math/rand/v2"
+	"strconv"
+	"testing"
+
+	"github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerList_FilterWorkersByLocalStorageState(t *testing.T) {
+	t.Run("list of workers with unhealthy workers", func(t *testing.T) {
+		workers := WorkerList([]*Worker{
+			NewWorker("global",
+				WithName("worker1"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(AvailableLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker2"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(LowStorageLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker3"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(CriticallyLowStorageLocalStorageState.String())),
+		})
+
+		healthyWorkers := FilterWorkersByLocalStorageState(workers)
+		assert.Equal(t, 1, len(healthyWorkers))
+	})
+
+	t.Run("list of workers with some unsupported worker versions", func(t *testing.T) {
+		workers := WorkerList([]*Worker{
+			NewWorker("global",
+				WithName("worker1"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(AvailableLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker2"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(UnknownLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker3"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(CriticallyLowStorageLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker4"),
+				WithReleaseVersion("Boundary v0.15.0"),
+				WithLocalStorageState(CriticallyLowStorageLocalStorageState.String())),
+		})
+
+		healthyWorkers := FilterWorkersByLocalStorageState(workers)
+		assert.Equal(t, 2, len(healthyWorkers))
+	})
+
+	t.Run("list of workers with no healthy workers", func(t *testing.T) {
+		workers := WorkerList([]*Worker{
+			NewWorker("global",
+				WithName("worker1"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(LowStorageLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker2"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(OutOfStorageLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker3"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(CriticallyLowStorageLocalStorageState.String())),
+		})
+
+		healthyWorkers := FilterWorkersByLocalStorageState(workers)
+		assert.Equal(t, 0, len(healthyWorkers))
+	})
+
+	t.Run("list of workers with all unknown local storage state", func(t *testing.T) {
+		workers := WorkerList([]*Worker{
+			NewWorker("global",
+				WithName("worker1"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(UnknownLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker2"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(UnknownLocalStorageState.String())),
+			NewWorker("global",
+				WithName("worker3"),
+				WithReleaseVersion("Boundary v0.16.0"),
+				WithLocalStorageState(UnknownLocalStorageState.String())),
+		})
+
+		healthyWorkers := FilterWorkersByLocalStorageState(workers)
+		assert.Equal(t, 3, len(healthyWorkers))
+	})
+}
+
+func Test_FilterStorageBucketCredentialByWriteAccess(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		name           string
+		sbcState       *plugin.StorageBucketCredentialState
+		expectedFilter bool
+	}{
+		{
+			name:           "nil sbcState",
+			expectedFilter: true,
+		},
+		{
+			name:           "nil underlying state",
+			sbcState:       &plugin.StorageBucketCredentialState{},
+			expectedFilter: true,
+		},
+		{
+			name: "nil write permission",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "ok state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Write: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_OK,
+					},
+				},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "unknown state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Write: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_UNKNOWN,
+					},
+				},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "error state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Write: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_ERROR,
+					},
+				},
+			},
+			expectedFilter: false,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert := assert.New(t)
+			assert.Equal(tc.expectedFilter, FilterStorageBucketCredentialByWriteAccess(tc.sbcState))
+		})
+	}
+}
+
+func Test_FilterStorageBucketCredentialByReadAccess(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		name           string
+		sbcState       *plugin.StorageBucketCredentialState
+		expectedFilter bool
+	}{
+		{
+			name:           "nil sbcState",
+			expectedFilter: true,
+		},
+		{
+			name:           "nil underlying state",
+			sbcState:       &plugin.StorageBucketCredentialState{},
+			expectedFilter: true,
+		},
+		{
+			name: "nil read permission",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "ok state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Read: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_OK,
+					},
+				},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "unknown state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Read: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_UNKNOWN,
+					},
+				},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "error state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Read: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_ERROR,
+					},
+				},
+			},
+			expectedFilter: false,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert := assert.New(t)
+			assert.Equal(tc.expectedFilter, FilterStorageBucketCredentialByReadAccess(tc.sbcState))
+		})
+	}
+}
+
+func Test_FilterStorageBucketCredentialByDeleteAccess(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		name           string
+		sbcState       *plugin.StorageBucketCredentialState
+		expectedFilter bool
+	}{
+		{
+			name:           "nil sbcState",
+			expectedFilter: true,
+		},
+		{
+			name:           "nil underlying state",
+			sbcState:       &plugin.StorageBucketCredentialState{},
+			expectedFilter: true,
+		},
+		{
+			name: "nil delete permission",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "ok state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Delete: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_OK,
+					},
+				},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "unknown state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Delete: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_UNKNOWN,
+					},
+				},
+			},
+			expectedFilter: true,
+		},
+		{
+			name: "error state",
+			sbcState: &plugin.StorageBucketCredentialState{
+				State: &plugin.Permissions{
+					Delete: &plugin.Permission{
+						State: plugin.StateType_STATE_TYPE_ERROR,
+					},
+				},
+			},
+			expectedFilter: false,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert := assert.New(t)
+			assert.Equal(tc.expectedFilter, FilterStorageBucketCredentialByDeleteAccess(tc.sbcState))
+		})
+	}
+}
+
+func TestShuffle(t *testing.T) {
+	t.Parallel()
+
+	t.Run("noElements", func(t *testing.T) {
+		in := WorkerList{}
+		out, err := in.Shuffle()
+		require.NoError(t, err)
+		require.Equal(t, in, out)
+	})
+
+	t.Run("oneElement", func(t *testing.T) {
+		in := WorkerList{NewWorker("test_scope", WithName("1"))}
+		out, err := in.Shuffle()
+		require.NoError(t, err)
+		require.Equal(t, in, out)
+	})
+
+	t.Run("multipleElements", func(t *testing.T) {
+		// We need a large amount of minimum workers here to statistically
+		// mitigate against the case where Shuffle just-so-happens to shuffle
+		// the elements into the same order they were in before.
+		n := rand.IntN(1000-99) + 100 // [100, 1000]
+
+		inOrder := make([]int, 0, n)
+		in := make(WorkerList, 0, n)
+		for i := 0; i < n; i++ {
+			inOrder = append(inOrder, i)
+			in = append(in, NewWorker("test", WithName(strconv.Itoa(i))))
+		}
+
+		out, err := in.Shuffle()
+		require.NoError(t, err)
+		require.ElementsMatch(t, in, out)
+
+		outOrder := make([]int, 0)
+		for _, w := range out {
+			i, _ := strconv.Atoi(w.Name)
+			outOrder = append(outOrder, i)
+		}
+		require.NotEqual(t, inOrder, outOrder)
+	})
+}

internal/server/worker_storage_bucket.go

@@ -18,6 +18,11 @@ import (
 
 const UpsertWorkerStorageBucketJobName = "upsert_worker_storage_bucket"
 
+type WorkerStorageBucket struct {
+	WorkerId string
+	Buckets  []*storagebuckets.StorageBucket
+}
+
 type UpdateStorageBucketCredential struct {
 	PublicId                  string `gorm:"primary_key"`
 	StorageBucketCredentialId string

internal/server/worker_tag.go

@@ -3,6 +3,11 @@
 
 package server
 
+import (
+	"encoding/json"
+	"fmt"
+)
+
 // A Tag is a custom key/value pair which can be attached to a Worker.
 // Multiple Tags may contain the same key and different values in which
 // case both key/value pairs are valid.  Tags can be sourced from either the
@@ -13,17 +18,48 @@ type Tag struct {
 	Value string
 }
 
-type TagSource string
+// Tags allows us to scan a JSON array of worker tags from the database
+type Tags []*Tag
 
-const (
-	ConfigurationTagSource TagSource = "configuration"
-	ApiTagSource           TagSource = "api"
-)
-
-func (t TagSource) isValid() bool {
-	return t == ConfigurationTagSource || t == ApiTagSource
+func (t *Tags) clone() Tags {
+	newTags := make(Tags, 0, len(*t))
+	for _, tag := range *t {
+		newTags = append(newTags, &Tag{Key: tag.Key, Value: tag.Value})
+	}
+	return newTags
 }
 
-func (t TagSource) String() string {
-	return string(t)
+// compactTags takes a list of Tags and returns a map of deduplicated, compated tags
+func compactTags(t ...*Tags) map[string][]string {
+	compactedTags := make(map[Tag]struct{})
+	for _, tags := range t {
+		for _, tag := range *tags {
+			compactedTags[*tag] = struct{}{}
+		}
+	}
+
+	tags := make(map[string][]string)
+	for t := range compactedTags {
+		tags[t.Key] = append(tags[t.Key], t.Value)
+	}
+	return tags
+}
+
+// Scan scans value into Tags, and implements the sql.Scanner interface
+func (t *Tags) Scan(in any) error {
+	var err error
+	switch v := in.(type) {
+	case string:
+		err = json.Unmarshal([]byte(v), &t)
+	case []byte:
+		err = json.Unmarshal(v, &t)
+	default:
+		return fmt.Errorf("cannot scan type %T into tags", in)
+	}
+	return err
+}
+
+// GormDataType gorm common data type (required)
+func (t *Tags) GormDataType() string {
+	return "tags"
 }

internal/server/worker_tag_test.go

@@ -0,0 +1,28 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package server
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDeduplicateTags(t *testing.T) {
+	foo := 5
+	tags := make([]*Tags, 0, foo)
+	for pippo := 0; pippo < foo; pippo++ {
+		tags = append(tags, &Tags{
+			{Key: "key", Value: "shared"},
+			{Key: "key", Value: fmt.Sprintf("unique-%d", pippo)},
+			{Key: fmt.Sprintf("key-%d", pippo), Value: fmt.Sprintf("another-unique-%d", pippo)},
+		})
+	}
+	gotMap := compactTags(tags...)
+	assert.Len(t, gotMap, 6)
+	assert.ElementsMatch(t, gotMap["key"], []string{"shared", "unique-0", "unique-1", "unique-2", "unique-3", "unique-4"})
+	assert.ElementsMatch(t, gotMap["key-0"], []string{"another-unique-0"})
+	assert.ElementsMatch(t, gotMap["key-4"], []string{"another-unique-4"})
+}

internal/server/worker_tags_test.go

@@ -25,49 +25,20 @@ func TestWorkerTags_Create(t *testing.T) {
 
 	tests := []struct {
 		name          string
-		want          *store.WorkerTag
+		want          *store.ApiTag
 		wantCreateErr bool
 	}{
 		{
-			name: "success api source",
-			want: &store.WorkerTag{
-				WorkerId: worker.GetPublicId(),
-				Key:      "key",
-				Value:    "value",
-				Source:   ApiTagSource.String(),
-			},
-		},
-		{
-			name: "success config source",
-			want: &store.WorkerTag{
-				WorkerId: worker.GetPublicId(),
-				Key:      "key",
-				Value:    "value",
-				Source:   ConfigurationTagSource.String(),
-			},
-		},
-		{
-			name: "unknown source",
-			want: &store.WorkerTag{
-				WorkerId: worker.GetPublicId(),
-				Key:      "key",
-				Value:    "value",
-				Source:   "unknown",
-			},
-			wantCreateErr: true,
-		},
-		{
-			name: "no source",
-			want: &store.WorkerTag{
+			name: "success- api tag",
+			want: &store.ApiTag{
 				WorkerId: worker.GetPublicId(),
 				Key:      "key",
 				Value:    "value",
 			},
-			wantCreateErr: true,
 		},
 		{
 			name: "bad worker id",
-			want: &store.WorkerTag{
+			want: &store.ApiTag{
 				WorkerId: "w_badworkeridthatdoesntexist",
 				Key:      "key",
 				Value:    "value",
@@ -76,7 +47,7 @@ func TestWorkerTags_Create(t *testing.T) {
 		},
 		{
 			name: "missing worker id",
-			want: &store.WorkerTag{
+			want: &store.ApiTag{
 				Key:   "key",
 				Value: "value",
 			},
@@ -84,7 +55,7 @@ func TestWorkerTags_Create(t *testing.T) {
 		},
 		{
 			name: "missing key",
-			want: &store.WorkerTag{
+			want: &store.ApiTag{
 				WorkerId: worker.GetPublicId(),
 				Value:    "value",
 			},
@@ -92,7 +63,7 @@ func TestWorkerTags_Create(t *testing.T) {
 		},
 		{
 			name: "missing value",
-			want: &store.WorkerTag{
+			want: &store.ApiTag{
 				WorkerId: worker.GetPublicId(),
 				Key:      "key",
 			},
@@ -111,6 +82,15 @@ func TestWorkerTags_Create(t *testing.T) {
 			}
 		})
 	}
+
+	// Create a config tag
+	configTag := &store.ConfigTag{
+		WorkerId: worker.GetPublicId(),
+		Key:      "key",
+		Value:    "value",
+	}
+	err := rw.Create(context.Background(), configTag)
+	assert.NoError(t, err)
 }
 
 func TestRepository_AddWorkerTags(t *testing.T) {
@@ -659,7 +639,7 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) {
 	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
 	require.NoError(err)
 	assert.Equal(uint32(2), worker.Version)
-	assert.Equal(len(manyTags), len(worker.apiTags))
+	assert.Equal(len(manyTags), len(worker.ApiTags))
 
 	// Delete a valid tag from worker
 	rowsDeleted, err = repo.DeleteWorkerTags(context.Background(), worker.PublicId, worker.Version, []*Tag{
@@ -670,8 +650,8 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) {
 	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
 	require.NoError(err)
 	assert.Equal(uint32(3), worker.Version)
-	assert.Contains(worker.apiTags, &Tag{Key: "key", Value: "value"})
-	assert.Contains(worker.apiTags, &Tag{Key: "key2", Value: "value2"})
+	assert.Contains(worker.ApiTags, &Tag{Key: "key", Value: "value"})
+	assert.Contains(worker.ApiTags, &Tag{Key: "key2", Value: "value2"})
 
 	// Add another valid tag to worker
 	_, err = repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, []*Tag{
@@ -681,8 +661,8 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) {
 	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
 	require.NoError(err)
 	assert.Equal(uint32(4), worker.Version)
-	assert.Contains(worker.apiTags, &Tag{Key: "key!", Value: "value!"})
-	assert.Equal(3, len(worker.apiTags))
+	assert.Contains(worker.ApiTags, &Tag{Key: "key!", Value: "value!"})
+	assert.Equal(3, len(worker.ApiTags))
 
 	// Set all tags to nil
 	set, err = repo.SetWorkerTags(context.Background(), worker.PublicId, worker.Version, nil)
@@ -691,8 +671,8 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) {
 	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
 	require.NoError(err)
 	assert.Equal(uint32(5), worker.Version)
-	assert.Equal(set, worker.apiTags)
-	assert.Equal(0, len(worker.apiTags))
+	assert.Equal(Tags(nil), worker.ApiTags)
+	assert.Equal(0, len(worker.ApiTags))
 
 	// Ensure config tags are untouched
 	for _, ct := range []*Tag{
@@ -701,9 +681,9 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) {
 		{Key: "key3", Value: "value3"},
 		{Key: "key?", Value: "value?"},
 	} {
-		assert.Contains(worker.configTags, ct)
+		assert.Contains(worker.ConfigTags, ct)
 	}
-	assert.Equal(4, len(worker.configTags))
+	assert.Equal(4, len(worker.ConfigTags))
 
 	// Go full circle
 	_, err = repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, manyTags)
@@ -711,8 +691,8 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) {
 	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
 	require.NoError(err)
 	assert.Equal(uint32(6), worker.Version)
-	assert.Equal(len(manyTags), len(worker.apiTags))
+	assert.Equal(len(manyTags), len(worker.ApiTags))
 	for _, t := range manyTags {
-		assert.Contains(worker.apiTags, t)
+		assert.Contains(worker.ApiTags, t)
 	}
 }

internal/server/worker_test.go

@@ -24,12 +24,12 @@ import (
 
 func TestWorkerCanonicalTags(t *testing.T) {
 	w := NewWorker(scope.Global.String())
-	w.apiTags = []*Tag{
+	w.ApiTags = []*Tag{
 		{Key: "key", Value: "apis unique"},
 		{Key: "key", Value: "shared"},
 		{Key: "key2", Value: "apis key2 unique"},
 	}
-	w.configTags = []*Tag{
+	w.ConfigTags = []*Tag{
 		{Key: "key", Value: "configs unique"},
 		{Key: "key", Value: "shared"},
 		{Key: "key3", Value: "configs key3 unique"},
@@ -48,12 +48,10 @@ func TestWorkerAggregate(t *testing.T) {
 	rw := db.New(conn)
 	ctx := context.Background()
 
-	getAggWorker := func(id string) *Worker {
-		agg := &workerAggregate{PublicId: id}
-		require.NoError(t, rw.LookupById(ctx, agg))
-		got, err := agg.toWorker(ctx)
-		assert.NoError(t, err)
-		return got
+	getWorker := func(id string) *Worker {
+		ret, err := lookupWorker(ctx, rw, id)
+		require.NoError(t, err)
+		return ret
 	}
 
 	t.Run("kms worker", func(t *testing.T) {
@@ -67,7 +65,7 @@ func TestWorkerAggregate(t *testing.T) {
 		w.PublicId = id
 		require.NoError(t, rw.Create(ctx, w))
 
-		got := getAggWorker(id)
+		got := getWorker(id)
 		assert.Equal(t, KmsWorkerType.String(), got.GetType())
 		assert.Equal(t, id, got.GetPublicId())
 		assert.Equal(t, scope.Global.String(), got.GetScopeId())
@@ -90,7 +88,7 @@ func TestWorkerAggregate(t *testing.T) {
 		w.PublicId = id
 		require.NoError(t, rw.Create(ctx, w))
 
-		got := getAggWorker(id)
+		got := getWorker(id)
 		assert.Equal(t, id, got.GetPublicId())
 		assert.Equal(t, uint32(1), got.GetVersion())
 		assert.NotNil(t, got.GetLastStatusTime())
@@ -109,20 +107,19 @@ func TestWorkerAggregate(t *testing.T) {
 		w.PublicId = id
 		require.NoError(t, rw.Create(ctx, w))
 		require.NoError(t, rw.Create(ctx,
-			&store.WorkerTag{
+			&store.ConfigTag{
 				WorkerId: id,
 				Key:      "key",
 				Value:    "val",
-				Source:   ConfigurationTagSource.String(),
 			}))
 
-		got := getAggWorker(id)
+		got := getWorker(id)
 		assert.Equal(t, id, got.GetPublicId())
 		assert.Equal(t, uint32(1), got.GetVersion())
 		assert.NotNil(t, got.GetLastStatusTime())
 		assert.NotNil(t, got.GetReleaseVersion())
-		assert.Empty(t, got.apiTags)
-		assert.Equal(t, got.configTags, []*Tag{{Key: "key", Value: "val"}})
+		assert.Empty(t, got.ApiTags)
+		assert.Equal(t, got.ConfigTags, Tags{{Key: "key", Value: "val"}})
 	})
 
 	t.Run("Worker with many config tag", func(t *testing.T) {
@@ -134,29 +131,26 @@ func TestWorkerAggregate(t *testing.T) {
 		w.Type = KmsWorkerType.String()
 		w.PublicId = id
 		require.NoError(t, rw.Create(ctx, w))
-		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+		require.NoError(t, rw.Create(ctx, &store.ConfigTag{
 			WorkerId: id,
 			Key:      "key",
 			Value:    "val",
-			Source:   ConfigurationTagSource.String(),
 		}))
-		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+		require.NoError(t, rw.Create(ctx, &store.ConfigTag{
 			WorkerId: id,
 			Key:      "key",
 			Value:    "val2",
-			Source:   ConfigurationTagSource.String(),
 		}))
-		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+		require.NoError(t, rw.Create(ctx, &store.ConfigTag{
 			WorkerId: id,
 			Key:      "key2",
 			Value:    "val2",
-			Source:   ConfigurationTagSource.String(),
 		}))
 
-		got := getAggWorker(id)
+		got := getWorker(id)
 		require.NotNil(t, got.GetLastStatusTime())
-		assert.Empty(t, got.apiTags)
-		assert.ElementsMatch(t, got.configTags, []*Tag{
+		assert.Empty(t, got.ApiTags)
+		assert.ElementsMatch(t, got.ConfigTags, []*Tag{
 			{Key: "key", Value: "val"},
 			{Key: "key", Value: "val2"},
 			{Key: "key2", Value: "val2"},
@@ -173,19 +167,18 @@ func TestWorkerAggregate(t *testing.T) {
 		w.PublicId = id
 		require.NoError(t, rw.Create(ctx, w))
 		require.NoError(t, rw.Create(ctx,
-			&store.WorkerTag{
+			&store.ApiTag{
 				WorkerId: id,
 				Key:      "key",
 				Value:    "val",
-				Source:   ApiTagSource.String(),
 			}))
 
-		got := getAggWorker(id)
+		got := getWorker(id)
 		assert.Equal(t, id, got.GetPublicId())
 		assert.Equal(t, uint32(1), got.GetVersion())
 		assert.NotNil(t, got.GetLastStatusTime())
 		assert.Empty(t, got.GetConfigTags())
-		assert.Equal(t, got.apiTags, []*Tag{{Key: "key", Value: "val"}})
+		assert.Equal(t, got.ApiTags, Tags{{Key: "key", Value: "val"}})
 	})
 
 	// Worker with mix of tag sources
@@ -198,32 +191,29 @@ func TestWorkerAggregate(t *testing.T) {
 		w.Type = KmsWorkerType.String()
 		w.PublicId = id
 		require.NoError(t, rw.Create(ctx, w))
-		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+		require.NoError(t, rw.Create(ctx, &store.ConfigTag{
 			WorkerId: id,
 			Key:      "key",
 			Value:    "val",
-			Source:   ConfigurationTagSource.String(),
 		}))
-		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+		require.NoError(t, rw.Create(ctx, &store.ApiTag{
 			WorkerId: id,
 			Key:      "key",
 			Value:    "val2",
-			Source:   ApiTagSource.String(),
 		}))
-		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+		require.NoError(t, rw.Create(ctx, &store.ApiTag{
 			WorkerId: id,
 			Key:      "key2",
 			Value:    "val2",
-			Source:   ApiTagSource.String(),
 		}))
 
-		got := getAggWorker(id)
+		got := getWorker(id)
 		require.NotNil(t, got.GetLastStatusTime())
-		assert.ElementsMatch(t, got.apiTags, []*Tag{
+		assert.ElementsMatch(t, got.ApiTags, []*Tag{
 			{Key: "key", Value: "val2"},
 			{Key: "key2", Value: "val2"},
 		})
-		assert.ElementsMatch(t, got.configTags, []*Tag{
+		assert.ElementsMatch(t, got.ConfigTags, []*Tag{
 			{Key: "key", Value: "val"},
 		})
 	})
@@ -239,7 +229,7 @@ func TestWorkerAggregate(t *testing.T) {
 		w.PublicId = id
 		require.NoError(t, rw.Create(ctx, w))
 
-		got := getAggWorker(id)
+		got := getWorker(id)
 		assert.Equal(t, UnknownLocalStorageState.String(), got.LocalStorageState)
 	})
 
@@ -255,7 +245,7 @@ func TestWorkerAggregate(t *testing.T) {
 		w.PublicId = id
 		require.NoError(t, rw.Create(ctx, w))
 
-		got := getAggWorker(id)
+		got := getWorker(id)
 		assert.Equal(t, AvailableLocalStorageState.String(), got.LocalStorageState)
 	})
 }