Downstream worker connections are tracked by worker id (#2949)

When possible downstream workers will include their worker id information in
the connections they establish with the upstream using the nodeenrollment
library.  The tracking listener now tracks connections that are connected in
this same way instead of just connections of type *tls.Conn.

internal/daemon/cluster/connstate.go

@@ -0,0 +1,61 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package cluster
+
+import (
+	"context"
+	"net"
+
+	"github.com/fatih/structs"
+	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/mitchellh/mapstructure"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+// WorkerConnectionInfo contains information about a worker that gets attached
+// to outgoing connections to an upstream worker or the controller through the
+// state of a *protocol.Conn.
+type WorkerConnectionInfo struct {
+	// The address that the dialing worker is dialing
+	UpstreamAddress string `mapstructure:"upstream_address,omitempty"`
+	// The worker id of the worker that is dialing
+	WorkerId string `mapstructure:"worker_id,omitempty"`
+	// The version of the worker that is dialing
+	BoundaryVersion string `mapstructure:"boundary_version,omitempty"`
+}
+
+// AsConnectionStateStruct builds a structPb in the format that is expected for
+// workers to attach their state to a *protocol.Conn through the withState option.
+func (w *WorkerConnectionInfo) AsConnectionStateStruct() (*structpb.Struct, error) {
+	if w == nil {
+		return nil, nil
+	}
+	s := structs.New(w)
+	s.TagName = "mapstructure"
+	return structpb.NewStruct(s.Map())
+}
+
+type stateProvider interface {
+	State() *structpb.Struct
+}
+
+// GetWorkerInfoFromStateMap returns the WorkerConnectionInfo for the provided
+// connection if the connection is a stateProvider.  If it is not a stateProvider
+// or if the provided state is nil, an error is returned.
+func GetWorkerInfoFromStateMap(ctx context.Context, c net.Conn) (*WorkerConnectionInfo, error) {
+	const op = "cluster.GetWorkerInfoFromStateMap"
+	var downstreamInfo WorkerConnectionInfo
+	pc, ok := c.(stateProvider)
+	if !ok {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "connection does not provide a State structpb.Struct")
+	}
+	st := pc.State()
+	if st == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "no state attached to connection")
+	}
+	if err := mapstructure.Decode(st.AsMap(), &downstreamInfo); err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	return &downstreamInfo, nil
+}

internal/daemon/cluster/connstate_test.go

@@ -0,0 +1,65 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package cluster
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+type testConn struct {
+	net.Conn
+	st *structpb.Struct
+}
+
+func (c *testConn) State() *structpb.Struct {
+	return c.st
+}
+
+func TestWorkerConnectionInfo(t *testing.T) {
+	connInfo := []*WorkerConnectionInfo{
+		{WorkerId: "something"},
+		{UpstreamAddress: "something"},
+		{BoundaryVersion: "something"},
+		{WorkerId: "something", UpstreamAddress: "something else", BoundaryVersion: "something yet more"},
+	}
+	for _, c := range connInfo {
+		st, err := c.AsConnectionStateStruct()
+		assert.NoError(t, err)
+
+		got, err := GetWorkerInfoFromStateMap(context.Background(), &testConn{st: st})
+		require.NoError(t, err)
+		assert.Equal(t, c, got)
+	}
+}
+
+func TestWorkerConnectionInfo_errors(t *testing.T) {
+	t.Run("wrong connection type", func(t *testing.T) {
+		c, _ := net.Pipe()
+		got, err := GetWorkerInfoFromStateMap(context.Background(), c)
+		assert.ErrorContains(t, err, "connection does not provide a State structpb.Struct")
+		assert.Nil(t, got)
+	})
+	t.Run("no state on connection", func(t *testing.T) {
+		got, err := GetWorkerInfoFromStateMap(context.Background(), &testConn{})
+		assert.ErrorContains(t, err, "no state attached")
+		assert.Nil(t, got)
+	})
+
+	t.Run("wrong type in struct", func(t *testing.T) {
+		st, err := structpb.NewStruct(map[string]interface{}{
+			"worker_id": 1.2480,
+		})
+		require.NoError(t, err)
+		require.NotNil(t, st)
+		got, err := GetWorkerInfoFromStateMap(context.Background(), &testConn{st: st})
+		assert.ErrorContains(t, err, "'worker_id' expected type 'string'")
+		assert.Nil(t, got)
+	})
+}

internal/daemon/cluster/downstreammanager.go

@@ -12,6 +12,10 @@ import (
 // connections to  a specific server.
 // It is safe to access DownstreamManager concurrently.
 type DownstreamManager struct {
+	keyToWorkerId    map[string]string
+	workerIdToKeyIds map[string]map[string]struct{}
+
+	// maps a key id to a connection initiated by a downstream worker
 	workerConnections map[string][]net.Conn
 	l                 sync.RWMutex
 }
@@ -19,9 +23,21 @@ type DownstreamManager struct {
 func NewDownstreamManager() *DownstreamManager {
 	return &DownstreamManager{
 		workerConnections: make(map[string][]net.Conn),
+		keyToWorkerId:     make(map[string]string),
+		workerIdToKeyIds:  make(map[string]map[string]struct{}),
 	}
 }
 
+func (m *DownstreamManager) mapKeyToWorkerId(keyId, wId string) {
+	m.l.Lock()
+	defer m.l.Unlock()
+	m.keyToWorkerId[keyId] = wId
+	if _, ok := m.workerIdToKeyIds[wId]; !ok {
+		m.workerIdToKeyIds[wId] = make(map[string]struct{})
+	}
+	m.workerIdToKeyIds[wId][keyId] = struct{}{}
+}
+
 // addConnection adds a connection associated with the provided downstream
 // worker key identifier.
 func (m *DownstreamManager) addConnection(id string, c net.Conn) {
@@ -30,41 +46,110 @@ func (m *DownstreamManager) addConnection(id string, c net.Conn) {
 	m.workerConnections[id] = append(m.workerConnections[id], c)
 }
 
-// Disconnect closes all connections associated with the provided worker key
+// disconnectWorkerId closes all connections associated with the provided
+// worker public id
+func (m *DownstreamManager) disconnectWorkerId(id string) {
+	m.l.Lock()
+	defer m.l.Unlock()
+	for k := range m.workerIdToKeyIds[id] {
+		for _, c := range m.workerConnections[k] {
+			c.Close()
+		}
+		delete(m.workerConnections, k)
+		delete(m.keyToWorkerId, k)
+	}
+	delete(m.workerIdToKeyIds, id)
+}
+
+// disconnectKeys closes all connections associated with the provided worker key
 // identifier.
-func (m *DownstreamManager) Disconnect(id string) {
+func (m *DownstreamManager) disconnectKeys(id string) {
 	m.l.Lock()
 	defer m.l.Unlock()
 	for _, c := range m.workerConnections[id] {
 		c.Close()
 	}
 	delete(m.workerConnections, id)
+
+	// clean up all the associations between worker id and key ids
+	if wId, ok := m.keyToWorkerId[id]; ok {
+		delete(m.keyToWorkerId, id)
+		delete(m.workerIdToKeyIds[wId], id)
+		if len(m.workerIdToKeyIds[wId]) == 0 {
+			delete(m.workerIdToKeyIds, wId)
+		}
+	}
 }
 
-// Connected returns a slice of worker key identifiers for all workers that are
-// currently being tracked by this downstream manager.
-func (m *DownstreamManager) Connected() []string {
+// Connected returns a connected state which provides the worker ids that are
+// being tracked and any key ids for which we don't know the worker id.
+func (m *DownstreamManager) Connected() *connectedState {
 	m.l.RLock()
 	defer m.l.RUnlock()
-	var r []string
+	seenWIds := make(map[string]struct{})
+	var workerIds, keyIds []string
 	for k, v := range m.workerConnections {
 		if len(v) > 0 {
-			r = append(r, k)
+			if w, ok := m.keyToWorkerId[k]; ok {
+				if _, ok := seenWIds[w]; !ok {
+					seenWIds[w] = struct{}{}
+					workerIds = append(workerIds, w)
+				}
+			} else {
+				keyIds = append(keyIds, k)
+			}
 		}
 	}
-	return r
+	return &connectedState{
+		dm:             m,
+		unmappedKeyIds: keyIds,
+		workerIds:      workerIds,
+	}
 }
 
-// DisconnectUnauthorized calls disconnects for all ids which are present in
-// the connected but not in the authorized slice of key ids.
-func DisconnectUnauthorized(dm *DownstreamManager, connected, authorized []string) {
-	am := make(map[string]struct{}, len(authorized))
-	for _, i := range authorized {
-		am[i] = struct{}{}
+// connectedState provides a state of the downstream manager and allows users
+// to request connections be disconnected that are capturred in this state.
+type connectedState struct {
+	dm             *DownstreamManager
+	unmappedKeyIds []string
+	workerIds      []string
+}
+
+// UnmappedKeyIds are the key ids for which no worker id association is known
+// and which are tracking at least 1 connection.
+func (s *connectedState) UnmappedKeyIds() []string {
+	return s.unmappedKeyIds
+}
+
+// WorkerIds are the public ids for workers which we are tracking connections.
+func (s *connectedState) WorkerIds() []string {
+	return s.workerIds
+}
+
+// DisconnectMissingWorkers disconnects all workers that are not in the slice
+// of worker ids provided but are tracked in this connected state.
+func (s *connectedState) DisconnectMissingWorkers(workers []string) {
+	aw := make(map[string]struct{}, len(workers))
+	for _, i := range workers {
+		aw[i] = struct{}{}
 	}
-	for _, i := range connected {
-		if _, found := am[i]; !found {
-			dm.Disconnect(i)
+	for _, w := range s.workerIds {
+		if _, found := aw[w]; !found {
+			s.dm.disconnectWorkerId(w)
+		}
+	}
+}
+
+// DisconnectMissingKeyIds disconnects all workers which are not in the slice of
+// key ids provided but are tracked in this connected state.
+func (s *connectedState) DisconnectMissingKeyIds(keyIds []string) {
+	am := make(map[string]struct{}, len(keyIds))
+	for _, i := range keyIds {
+		am[i] = struct{}{}
+	}
+	for _, i := range s.unmappedKeyIds {
+		if _, found := am[i]; !found {
+			s.dm.disconnectKeys(i)
 		}
 	}
 }

internal/daemon/cluster/downstreammanager_test.go

@@ -15,86 +15,339 @@ import (
 
 func TestNewDownstreamManager(t *testing.T) {
 	dm := NewDownstreamManager()
-	assert.Empty(t, dm.Connected())
+	assert.Empty(t, dm.Connected().UnmappedKeyIds())
+	assert.Empty(t, dm.Connected().WorkerIds())
 
 	t.Run("single connection", func(t *testing.T) {
 		w1, g1 := net.Pipe()
 		dm.addConnection("1", w1)
-		assert.Equal(t, []string{"1"}, dm.Connected())
+		st := dm.Connected()
+		assert.Equal(t, []string{"1"}, st.UnmappedKeyIds())
 
-		dm.Disconnect("1")
+		dm.disconnectKeys("1")
 		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 		assert.ErrorIs(t, g1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 
-		assert.Empty(t, dm.Connected())
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+	})
+
+	t.Run("single connection with worker", func(t *testing.T) {
+		w1, g1 := net.Pipe()
+		dm.addConnection("1", w1)
+		dm.mapKeyToWorkerId("1", "w1")
+
+		st := dm.Connected()
+		assert.Empty(t, st.UnmappedKeyIds())
+		assert.Equal(t, []string{"w1"}, st.WorkerIds())
+
+		dm.disconnectWorkerId("w1")
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+		assert.Empty(t, dm.Connected().WorkerIds())
 	})
 
 	t.Run("multiple connection with single name", func(t *testing.T) {
 		w1, g1 := net.Pipe()
 		dm.addConnection("1", w1)
-		assert.Equal(t, []string{"1"}, dm.Connected())
+		assert.Equal(t, []string{"1"}, dm.Connected().UnmappedKeyIds())
 		w2, g2 := net.Pipe()
 		dm.addConnection("1", w2)
-		assert.Equal(t, []string{"1"}, dm.Connected())
+		assert.Equal(t, []string{"1"}, dm.Connected().UnmappedKeyIds())
 
 		assert.NoError(t, w1.SetReadDeadline(time.Now()))
 		assert.NoError(t, g1.SetReadDeadline(time.Now()))
 		assert.NoError(t, w2.SetReadDeadline(time.Now()))
 		assert.NoError(t, g2.SetReadDeadline(time.Now()))
 
-		dm.Disconnect("1")
+		dm.disconnectKeys("1")
 		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 		assert.ErrorIs(t, g1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 		assert.ErrorIs(t, w2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 		assert.ErrorIs(t, g2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 
-		assert.Empty(t, dm.Connected())
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
 	})
 
-	t.Run("multiple connection with different names", func(t *testing.T) {
+	t.Run("multiple connection with single key mapped to worker", func(t *testing.T) {
 		w1, g1 := net.Pipe()
 		dm.addConnection("1", w1)
-		assert.Equal(t, []string{"1"}, dm.Connected())
+		assert.Equal(t, []string{"1"}, dm.Connected().UnmappedKeyIds())
+		w2, g2 := net.Pipe()
+		dm.addConnection("1", w2)
+		assert.Equal(t, []string{"1"}, dm.Connected().UnmappedKeyIds())
+
+		dm.mapKeyToWorkerId("1", "w1")
+		assert.Equal(t, []string{"w1"}, dm.Connected().WorkerIds())
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, g1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2.SetReadDeadline(time.Now()))
+		assert.NoError(t, g2.SetReadDeadline(time.Now()))
+
+		dm.disconnectWorkerId("w1")
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+		assert.Empty(t, dm.Connected().WorkerIds())
+	})
+
+	t.Run("multiple connection with different keys", func(t *testing.T) {
+		w1, g1 := net.Pipe()
+		dm.addConnection("1", w1)
+		assert.Equal(t, []string{"1"}, dm.Connected().UnmappedKeyIds())
 		w2, g2 := net.Pipe()
 		dm.addConnection("2", w2)
-		got := dm.Connected()
+		got := dm.Connected().UnmappedKeyIds()
 		sort.Strings(got)
 		assert.Equal(t, []string{"1", "2"}, got)
 
-		dm.Disconnect("1")
+		dm.disconnectKeys("1")
 		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 		assert.ErrorIs(t, g1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 
 		assert.NoError(t, w2.SetReadDeadline(time.Now()))
 		assert.NoError(t, g2.SetReadDeadline(time.Now()))
 
-		assert.Equal(t, []string{"2"}, dm.Connected())
+		assert.Equal(t, []string{"2"}, dm.Connected().UnmappedKeyIds())
 
-		dm.Disconnect("2")
+		dm.disconnectKeys("2")
 		assert.ErrorIs(t, w2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
 		assert.ErrorIs(t, g2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+		assert.Empty(t, dm.Connected().WorkerIds())
+	})
+
+	t.Run("multiple connection with different keys to same worker", func(t *testing.T) {
+		w1, g1 := net.Pipe()
+		dm.addConnection("1", w1)
+		assert.Equal(t, []string{"1"}, dm.Connected().UnmappedKeyIds())
+		w2, g2 := net.Pipe()
+		dm.addConnection("2", w2)
+		assert.ElementsMatch(t, []string{"1", "2"}, dm.Connected().UnmappedKeyIds())
+
+		dm.mapKeyToWorkerId("1", "w1")
+		dm.mapKeyToWorkerId("2", "w1")
+		assert.Equal(t, []string{"w1"}, dm.Connected().WorkerIds())
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, g1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2.SetReadDeadline(time.Now()))
+		assert.NoError(t, g2.SetReadDeadline(time.Now()))
+
+		dm.disconnectWorkerId("w1")
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+		assert.Empty(t, dm.Connected().WorkerIds())
+	})
+
+	t.Run("multiple connection with different key to partial worker", func(t *testing.T) {
+		w1, g1 := net.Pipe()
+		dm.addConnection("1", w1)
+		assert.Equal(t, []string{"1"}, dm.Connected().UnmappedKeyIds())
+		w2, g2 := net.Pipe()
+		dm.addConnection("2", w2)
+		assert.ElementsMatch(t, []string{"1", "2"}, dm.Connected().UnmappedKeyIds())
+
+		dm.mapKeyToWorkerId("1", "w1")
+		assert.Equal(t, []string{"w1"}, dm.Connected().WorkerIds())
+		assert.Equal(t, []string{"2"}, dm.Connected().UnmappedKeyIds())
+
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, g1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2.SetReadDeadline(time.Now()))
+		assert.NoError(t, g2.SetReadDeadline(time.Now()))
+
+		dm.disconnectWorkerId("w1")
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.NoError(t, w2.SetReadDeadline(time.Now()))
+		assert.NoError(t, g2.SetReadDeadline(time.Now()))
+		dm.disconnectKeys("2")
+		assert.ErrorIs(t, w2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+		assert.Empty(t, dm.Connected().WorkerIds())
+	})
+
+	t.Run("multiple connection with different key to different workers", func(t *testing.T) {
+		w1, g1 := net.Pipe()
+		dm.addConnection("1", w1)
+		assert.Equal(t, []string{"1"}, dm.Connected().UnmappedKeyIds())
+		w2, g2 := net.Pipe()
+		dm.addConnection("2", w2)
+		assert.ElementsMatch(t, []string{"1", "2"}, dm.Connected().UnmappedKeyIds())
+
+		dm.mapKeyToWorkerId("1", "w1")
+		dm.mapKeyToWorkerId("2", "w2")
+		assert.Equal(t, []string{"w1", "w2"}, dm.Connected().WorkerIds())
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, g1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2.SetReadDeadline(time.Now()))
+		assert.NoError(t, g2.SetReadDeadline(time.Now()))
+
+		dm.disconnectWorkerId("w1")
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.NoError(t, w2.SetReadDeadline(time.Now()))
+		assert.NoError(t, g2.SetReadDeadline(time.Now()))
+		dm.disconnectWorkerId("w2")
+		assert.ErrorIs(t, w2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, g2.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.Empty(t, dm.Connected().UnmappedKeyIds())
+		assert.Empty(t, dm.Connected().WorkerIds())
 	})
 }
 
-func TestDisconnectUnauthorized(t *testing.T) {
-	dm := NewDownstreamManager()
-	w1, _ := net.Pipe()
-	dm.addConnection("w1", w1)
-	w2a, _ := net.Pipe()
-	dm.addConnection("w2", w2a)
-	w2b, _ := net.Pipe()
-	dm.addConnection("w2", w2b)
-	w3, _ := net.Pipe()
-	dm.addConnection("w3", w3)
+func TestConnectionState_Disconnect(t *testing.T) {
+	t.Run("disconnect some key ids", func(t *testing.T) {
+		dm := NewDownstreamManager()
+		w1, _ := net.Pipe()
+		dm.addConnection("1", w1)
+		w2a, _ := net.Pipe()
+		dm.addConnection("2", w2a)
+		w2b, _ := net.Pipe()
+		dm.addConnection("2", w2b)
+		w3, _ := net.Pipe()
+		dm.addConnection("3", w3)
 
-	assert.NoError(t, w1.SetReadDeadline(time.Now()))
-	assert.NoError(t, w2a.SetReadDeadline(time.Now()))
-	assert.NoError(t, w2b.SetReadDeadline(time.Now()))
-	assert.NoError(t, w3.SetReadDeadline(time.Now()))
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2a.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2b.SetReadDeadline(time.Now()))
+		assert.NoError(t, w3.SetReadDeadline(time.Now()))
 
-	DisconnectUnauthorized(dm, dm.Connected(), []string{"w3"})
-	assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
-	assert.ErrorIs(t, w2a.SetReadDeadline(time.Now()), io.ErrClosedPipe)
-	assert.ErrorIs(t, w2b.SetReadDeadline(time.Now()), io.ErrClosedPipe)
-	assert.NoError(t, w3.SetReadDeadline(time.Now()))
+		st := dm.Connected()
+		assert.ElementsMatch(t, []string{"1", "2", "3"}, st.UnmappedKeyIds())
+
+		st.DisconnectMissingKeyIds([]string{"3"})
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2a.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2b.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.NoError(t, w3.SetReadDeadline(time.Now()))
+	})
+
+	t.Run("disconnect all key ids", func(t *testing.T) {
+		dm := NewDownstreamManager()
+		w1, _ := net.Pipe()
+		dm.addConnection("1", w1)
+		w2a, _ := net.Pipe()
+		dm.addConnection("2", w2a)
+		w2b, _ := net.Pipe()
+		dm.addConnection("2", w2b)
+		w3, _ := net.Pipe()
+		dm.addConnection("3", w3)
+
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2a.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2b.SetReadDeadline(time.Now()))
+		assert.NoError(t, w3.SetReadDeadline(time.Now()))
+
+		st := dm.Connected()
+		assert.ElementsMatch(t, []string{"1", "2", "3"}, st.UnmappedKeyIds())
+
+		st.DisconnectMissingKeyIds(nil)
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2a.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2b.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w3.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+	})
+
+	t.Run("disconnect 1 worker id leave unmapped key id alone", func(t *testing.T) {
+		dm := NewDownstreamManager()
+		w1, _ := net.Pipe()
+		dm.addConnection("1", w1)
+		w2a, _ := net.Pipe()
+		dm.addConnection("2", w2a)
+		w2b, _ := net.Pipe()
+		dm.addConnection("2", w2b)
+		w3, _ := net.Pipe()
+		dm.addConnection("3", w3)
+
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2a.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2b.SetReadDeadline(time.Now()))
+		assert.NoError(t, w3.SetReadDeadline(time.Now()))
+
+		dm.mapKeyToWorkerId("1", "w1")
+		dm.mapKeyToWorkerId("2", "w1")
+		st := dm.Connected()
+		assert.ElementsMatch(t, []string{"3"}, st.UnmappedKeyIds())
+		assert.ElementsMatch(t, []string{"w1"}, st.WorkerIds())
+
+		st.DisconnectMissingWorkers(nil)
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2a.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2b.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.NoError(t, w3.SetReadDeadline(time.Now()))
+	})
+
+	t.Run("disconnect 1 key id leave worker id alone", func(t *testing.T) {
+		dm := NewDownstreamManager()
+		w1, _ := net.Pipe()
+		dm.addConnection("1", w1)
+		w2a, _ := net.Pipe()
+		dm.addConnection("2", w2a)
+		w2b, _ := net.Pipe()
+		dm.addConnection("2", w2b)
+		w3, _ := net.Pipe()
+		dm.addConnection("3", w3)
+
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2a.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2b.SetReadDeadline(time.Now()))
+		assert.NoError(t, w3.SetReadDeadline(time.Now()))
+
+		dm.mapKeyToWorkerId("1", "w1")
+		dm.mapKeyToWorkerId("2", "w1")
+		st := dm.Connected()
+		assert.ElementsMatch(t, []string{"3"}, st.UnmappedKeyIds())
+		assert.ElementsMatch(t, []string{"w1"}, st.WorkerIds())
+
+		st.DisconnectMissingKeyIds(nil)
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2a.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2b.SetReadDeadline(time.Now()))
+		assert.ErrorIs(t, w3.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+	})
+
+	t.Run("disconnect 1 worker id prividing a different worker", func(t *testing.T) {
+		dm := NewDownstreamManager()
+		w1, _ := net.Pipe()
+		dm.addConnection("1", w1)
+		w2a, _ := net.Pipe()
+		dm.addConnection("2", w2a)
+		w2b, _ := net.Pipe()
+		dm.addConnection("2", w2b)
+		w3, _ := net.Pipe()
+		dm.addConnection("3", w3)
+
+		assert.NoError(t, w1.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2a.SetReadDeadline(time.Now()))
+		assert.NoError(t, w2b.SetReadDeadline(time.Now()))
+		assert.NoError(t, w3.SetReadDeadline(time.Now()))
+
+		dm.mapKeyToWorkerId("1", "w1")
+		dm.mapKeyToWorkerId("2", "w1")
+		dm.mapKeyToWorkerId("3", "w2")
+		st := dm.Connected()
+		assert.Empty(t, st.UnmappedKeyIds())
+		assert.ElementsMatch(t, []string{"w1", "w2"}, st.WorkerIds())
+
+		st.DisconnectMissingWorkers([]string{"w2"})
+		assert.ErrorIs(t, w1.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2a.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.ErrorIs(t, w2b.SetReadDeadline(time.Now()), io.ErrClosedPipe)
+		assert.NoError(t, w3.SetReadDeadline(time.Now()))
+	})
 }

internal/daemon/cluster/handlers/worker_service.go

@@ -166,21 +166,34 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 		responseControllers = append(responseControllers, thisController)
 	}
 
-	workerAuthRepo, err := ws.workerAuthRepoFn()
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error getting worker auth repo"))
-		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to lookup worker auth info: %v", err)
+	authorizedWorkerList := &pbs.AuthorizedWorkerList{}
+	if len(req.GetConnectedWorkerPublicIds()) > 0 {
+		knownConnectedWorkers, err := serverRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithWorkerPool(req.GetConnectedWorkerPublicIds()), server.WithLiveness(-1))
+		if err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error getting known connected worker ids"))
+			return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error getting known connected worker ids: %v", err)
+		}
+		authorizedWorkerList.WorkerPublicIds = dcommon.WorkerList(knownConnectedWorkers).PublicIds()
 	}
-	authorizedWorkers, err := workerAuthRepo.FilterToAuthorizedWorkerKeyIds(ctx, req.GetConnectedWorkerKeyIdentifiers())
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error getting authorizable worker key ids"))
-		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error getting authorized worker key ids: %v", err)
+
+	if len(req.GetConnectedWorkerKeyIdentifiers()) > 0 {
+		workerAuthRepo, err := ws.workerAuthRepoFn()
+		if err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error getting worker auth repo"))
+			return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to lookup worker auth info: %v", err)
+		}
+		authorizedKeyIds, err := workerAuthRepo.FilterToAuthorizedWorkerKeyIds(ctx, req.GetConnectedWorkerKeyIdentifiers())
+		if err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error getting authorized worker key ids"))
+			return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error getting authorized worker key ids: %v", err)
+		}
+		authorizedWorkerList.WorkerKeyIdentifiers = authorizedKeyIds
 	}
 
 	ret := &pbs.StatusResponse{
 		CalculatedUpstreams: responseControllers,
 		WorkerId:            wrk.GetPublicId(),
-		AuthorizedWorkers:   &pbs.AuthorizedWorkerList{WorkerKeyIdentifiers: authorizedWorkers},
+		AuthorizedWorkers:   authorizedWorkerList,
 	}
 
 	stateReport := make([]*session.StateReport, 0, len(req.GetJobs()))

internal/daemon/cluster/handlers/worker_service_status_test.go

@@ -782,8 +782,8 @@ func TestStatusAuthorizedWorkers(t *testing.T) {
 
 	worker1 := server.TestKmsWorker(t, conn, wrapper)
 	var w1KeyId, w2KeyId string
-	_ = server.TestPkiWorker(t, conn, wrapper, server.WithTestPkiWorkerAuthorizedKeyId(&w1KeyId))
-	_ = server.TestPkiWorker(t, conn, wrapper, server.WithTestPkiWorkerAuthorizedKeyId(&w2KeyId))
+	w1 := server.TestPkiWorker(t, conn, wrapper, server.WithTestPkiWorkerAuthorizedKeyId(&w1KeyId))
+	w2 := server.TestPkiWorker(t, conn, wrapper, server.WithTestPkiWorkerAuthorizedKeyId(&w2KeyId))
 
 	s := NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, nil, new(sync.Map), kmsCache, new(atomic.Int64))
 	require.NotNil(t, s)
@@ -862,6 +862,57 @@ func TestStatusAuthorizedWorkers(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:    "Some authorized connected downstreams with worker public ids",
+			wantErr: false,
+			req: &pbs.StatusRequest{
+				WorkerStatus: &pb.ServerWorkerStatus{
+					PublicId: worker1.GetPublicId(),
+					Name:     worker1.GetName(),
+					Address:  worker1.GetAddress(),
+				},
+				ConnectedWorkerKeyIdentifiers: []string{w2KeyId, "unknown"},
+				ConnectedWorkerPublicIds:      []string{w1.GetPublicId(), "unknown"},
+			},
+			want: &pbs.StatusResponse{
+				CalculatedUpstreams: []*pbs.UpstreamServer{
+					{
+						Type:    pbs.UpstreamServer_TYPE_CONTROLLER,
+						Address: "127.0.0.1",
+					},
+				},
+				WorkerId: worker1.PublicId,
+				AuthorizedWorkers: &pbs.AuthorizedWorkerList{
+					WorkerKeyIdentifiers: []string{w2KeyId},
+					WorkerPublicIds:      []string{w1.GetPublicId()},
+				},
+			},
+		},
+		{
+			name:    "Some authorized connected downstreams with only worker public ids",
+			wantErr: false,
+			req: &pbs.StatusRequest{
+				WorkerStatus: &pb.ServerWorkerStatus{
+					PublicId: worker1.GetPublicId(),
+					Name:     worker1.GetName(),
+					Address:  worker1.GetAddress(),
+				},
+				ConnectedWorkerKeyIdentifiers: []string{"unknown"},
+				ConnectedWorkerPublicIds:      []string{w1.GetPublicId(), w2.GetPublicId(), "unknown"},
+			},
+			want: &pbs.StatusResponse{
+				CalculatedUpstreams: []*pbs.UpstreamServer{
+					{
+						Type:    pbs.UpstreamServer_TYPE_CONTROLLER,
+						Address: "127.0.0.1",
+					},
+				},
+				WorkerId: worker1.PublicId,
+				AuthorizedWorkers: &pbs.AuthorizedWorkerList{
+					WorkerPublicIds: []string{w1.GetPublicId(), w2.GetPublicId()},
+				},
+			},
+		},
 	}
 
 	for _, tc := range cases {
@@ -877,6 +928,8 @@ func TestStatusAuthorizedWorkers(t *testing.T) {
 			}
 			sort.Strings(got.GetAuthorizedWorkers().GetWorkerKeyIdentifiers())
 			sort.Strings(tc.want.GetAuthorizedWorkers().GetWorkerKeyIdentifiers())
+			sort.Strings(got.GetAuthorizedWorkers().GetWorkerPublicIds())
+			sort.Strings(tc.want.GetAuthorizedWorkers().GetWorkerPublicIds())
 			assert.Empty(
 				cmp.Diff(
 					tc.want,

internal/daemon/cluster/tracking_listener.go

@@ -6,26 +6,32 @@ package cluster
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
 	"net"
 
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	nodee "github.com/hashicorp/nodeenrollment"
+	"github.com/hashicorp/nodeenrollment/protocol"
 )
 
+const sourceEventInfoString = "tracking listener source"
+
 // trackingListener tracks a nodee connection in a DownstreamManager when a
 // worker has connected successfully.
 type trackingListener struct {
-	ctx context.Context
-	ln  net.Listener
-	dsm *DownstreamManager
+	ctx    context.Context
+	ln     net.Listener
+	dsm    *DownstreamManager
+	source string
 }
 
 // NewTrackingListener returns a listener which adds all connections made
 // through it to the provided DownstreamManager.  The net.Conn returned by
 // Accept is only tracked if it is a *tls.Conn and was created by the
-// nodeenrollment library.
-func NewTrackingListener(ctx context.Context, l net.Listener, m *DownstreamManager) (net.Listener, error) {
+// nodeenrollment library. The source string is provided to add context to
+// logs output by the NewTrackingListener.
+func NewTrackingListener(ctx context.Context, l net.Listener, m *DownstreamManager, source string) (net.Listener, error) {
 	const op = "common.NewTrackingListener"
 	switch {
 	case m == nil:
@@ -34,13 +40,14 @@ func NewTrackingListener(ctx context.Context, l net.Listener, m *DownstreamManag
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "nil base listener")
 	}
 	return &trackingListener{
-		ctx: ctx,
-		ln:  l,
-		dsm: m,
+		ctx:    ctx,
+		ln:     l,
+		dsm:    m,
+		source: source,
 	}, nil
 }
 
-// Accept satisfies the net.Listener interface.  If the the wrapped listener
+// Accept satisfies the net.Listener interface.  If the wrapped listener
 // must return a tls.Conn or an error is returned.  If the tls.Conn has no
 // PeerCertificates then no error is returned and the Conn is not added to the
 // DownstreamManager.
@@ -51,26 +58,50 @@ func (e *trackingListener) Accept() (net.Conn, error) {
 		return conn, err
 	}
 
-	tlsConn, ok := conn.(*tls.Conn)
-	if !ok {
-		return conn, err
+	eventingOpts := []any{sourceEventInfoString, e.source}
+	var workerId string
+	var tlsConn *tls.Conn
+	switch c := conn.(type) {
+	case *protocol.Conn:
+		tlsConn = c.Conn
+		wi, err := GetWorkerInfoFromStateMap(e.ctx, c)
+		if err != nil || wi == nil || len(wi.WorkerId) == 0 {
+			event.WriteSysEvent(e.ctx, op, "did not get worker information from state map")
+			break
+		}
+		workerId = wi.WorkerId
+		eventingOpts = append(eventingOpts, "worker id", workerId)
+	case *tls.Conn:
+		tlsConn = c
+	default:
+		event.WriteError(e.ctx, op, fmt.Errorf("unexpected connection type: %T", c), event.WithInfo(eventingOpts...))
+		return conn, nil
 	}
 
-	if tlsConn == nil || len(tlsConn.ConnectionState().PeerCertificates) == 0 {
+	if tlsConn == nil {
+		event.WriteError(e.ctx, op, fmt.Errorf("nil tlsConn"), event.WithInfo(eventingOpts...))
+		return conn, nil
+	}
+	if len(tlsConn.ConnectionState().PeerCertificates) == 0 {
+		event.WriteError(e.ctx, op, fmt.Errorf("tls conn without any peer certificates"), event.WithInfo(eventingOpts...))
 		return conn, nil
 	}
 
 	keyId, err := nodee.KeyIdFromPkix(tlsConn.ConnectionState().PeerCertificates[0].SubjectKeyId)
 	if err != nil {
-		// Create an error so it gets written out to the event log
-		_ = errors.Wrap(e.ctx, err, op)
+		event.WriteError(e.ctx, op, err, event.WithInfo(sourceEventInfoString, e.source))
 	}
 	if keyId == "" {
 		// No key id means there is nothing to track.
+		event.WriteError(e.ctx, op, fmt.Errorf("no key id found for connection"), event.WithInfo(eventingOpts...))
 		return conn, nil
 	}
+	eventingOpts = append(eventingOpts, "key_id", keyId)
+	if len(workerId) > 0 {
+		e.dsm.mapKeyToWorkerId(keyId, workerId)
+	}
 	e.dsm.addConnection(keyId, conn)
-	event.WriteSysEvent(e.ctx, op, "tracking worker connection", "key_id", keyId)
+	event.WriteSysEvent(e.ctx, op, "tracking worker connection", eventingOpts...)
 
 	return conn, nil
 }

internal/daemon/common/eventing_listener.go

@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	nodee "github.com/hashicorp/nodeenrollment"
+	"github.com/hashicorp/nodeenrollment/protocol"
 )
 
 // EventingListener simply sends an event when a worker has connected
@@ -41,9 +42,14 @@ func (e *eventingListener) Accept() (net.Conn, error) {
 		return conn, err
 	}
 
-	tlsConn, ok := conn.(*tls.Conn)
-	if !ok {
-		return conn, err
+	var tlsConn *tls.Conn
+	switch c := conn.(type) {
+	case *tls.Conn:
+		tlsConn = c
+	case *protocol.Conn:
+		tlsConn = c.Conn
+	default:
+		return conn, nil
 	}
 
 	if tlsConn != nil && len(tlsConn.ConnectionState().PeerCertificates) > 0 {

internal/daemon/controller/listeners.go

@@ -21,11 +21,18 @@ import (
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/go-multierror"
+	nodee "github.com/hashicorp/nodeenrollment"
 	nodeenet "github.com/hashicorp/nodeenrollment/net"
 	"github.com/hashicorp/nodeenrollment/protocol"
 	"google.golang.org/grpc"
 )
 
+const (
+	// the purpose strings used to identify listeners
+	reverseGrpcListenerPurpose = "reverse-grpc"
+	grpcListenerPurpose        = "grpc"
+)
+
 // the function that handles a secondary connection over a provided listener
 var handleSecondaryConnection = closeListener
 
@@ -151,7 +158,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 	// This handles connections coming in on the cluster port that are
 	// authenticated via nodeenrollment but not with any extra purpose; these
 	// are normal worker connections
-	nodeeAuthedListener, err := splitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto)
+	nodeeAuthedListener, err := splitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto, nodee.WithNativeConns(true))
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating node enrollment authed split listener: %w", err)
 	}
@@ -164,7 +171,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 
 	// This wraps the normal pki worker connections with a listener which adds
 	// the worker key id of the connections to the controller's pkiConnManager.
-	pkiWorkerTrackingListener, err := cluster.NewTrackingListener(c.baseContext, eventingAuthedListener, c.pkiConnManager)
+	pkiWorkerTrackingListener, err := cluster.NewTrackingListener(c.baseContext, eventingAuthedListener, c.pkiConnManager, sourcePurpose(grpcListenerPurpose))
 	if err != nil {
 		return nil, fmt.Errorf("%s: error creating pki worker tracking listener: %w", op, err)
 	}
@@ -186,14 +193,14 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 
 	// Connections coming in here are authed by nodeenrollment and are for the
 	// reverse grpc purpose
-	reverseGrpcListener, err := splitListener.GetListener(common.ReverseGrpcConnectionAlpnValue)
+	reverseGrpcListener, err := splitListener.GetListener(common.ReverseGrpcConnectionAlpnValue, nodee.WithNativeConns(true))
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating reverse gprc connection split listener: %w", err)
 	}
 
 	// This wraps the reverse grpc pki worker connections with a listener which
 	// adds the worker key id of the connections to the pkiConnManager.
-	revPkiWorkerTrackingListener, err := cluster.NewTrackingListener(c.baseContext, reverseGrpcListener, c.pkiConnManager)
+	revPkiWorkerTrackingListener, err := cluster.NewTrackingListener(c.baseContext, reverseGrpcListener, c.pkiConnManager, sourcePurpose(reverseGrpcListenerPurpose))
 	if err != nil {
 		return nil, fmt.Errorf("%s: error creating reverse grpc pki worker tracking listener: %w", op, err)
 	}
@@ -246,7 +253,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 	ln.GrpcServer = workerServer
 
 	return func() {
-		err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(multiplexingReverseGrpcListener, "reverse-grpc"),
+		err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(multiplexingReverseGrpcListener, reverseGrpcListenerPurpose),
 			c.downstreamRoutes)
 		if err != nil {
 			event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error"))
@@ -258,7 +265,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 			}
 		}()
 		go func() {
-			err := ln.GrpcServer.Serve(metric.InstrumentClusterTrackingListener(multiplexingAuthedListener, "grpc"))
+			err := ln.GrpcServer.Serve(metric.InstrumentClusterTrackingListener(multiplexingAuthedListener, grpcListenerPurpose))
 			if err != nil {
 				event.WriteError(c.baseContext, op, err, event.WithInfoMsg("multiplexingAuthedListener error"))
 			}
@@ -372,3 +379,7 @@ func listenerCloseErrorCheck(lnType string, err error) error {
 
 	return err
 }
+
+func sourcePurpose(purpose string) string {
+	return fmt.Sprintf("controller %s", purpose)
+}

internal/daemon/controller/tickers.go

@@ -10,10 +10,12 @@ import (
 	"time"
 
 	"github.com/hashicorp/boundary/internal/daemon/cluster"
-	"github.com/hashicorp/boundary/internal/server/store"
-
+	"github.com/hashicorp/boundary/internal/daemon/common"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/server"
+	"github.com/hashicorp/boundary/internal/server/store"
+	"github.com/hashicorp/boundary/internal/types/scope"
 )
 
 // In the future we could make this configurable
@@ -190,18 +192,34 @@ func (c *Controller) startWorkerConnectionMaintenanceTicking(cancelCtx context.C
 				return
 
 			case <-timer.C:
-				repo, err := c.WorkerAuthRepoStorageFn()
-				if err != nil {
-					event.WriteError(cancelCtx, op, err, event.WithInfoMsg("error fetching repository for cluster connection maintenance"))
-					break
+				connectionState := m.Connected()
+				if len(connectionState.WorkerIds()) > 0 {
+					serverRepo, err := c.ServersRepoFn()
+					if err != nil {
+						event.WriteError(cancelCtx, op, err, event.WithInfoMsg("error fetching server repository for cluster connection maintenance"))
+						break
+					}
+					knownWorker, err := serverRepo.ListWorkers(cancelCtx, []string{scope.Global.String()}, server.WithWorkerPool(connectionState.WorkerIds()), server.WithLiveness(-1))
+					if err != nil {
+						event.WriteError(cancelCtx, op, err, event.WithInfoMsg("couldn't get known workers from repo"))
+						break
+					}
+					connectionState.DisconnectMissingWorkers(common.WorkerList(knownWorker).PublicIds())
 				}
-				wKeyIds := m.Connected()
-				authorized, err := repo.FilterToAuthorizedWorkerKeyIds(cancelCtx, wKeyIds)
-				if err != nil {
-					event.WriteError(cancelCtx, op, err, event.WithInfoMsg("couldn't get authorized workers from repo"))
-					break
+
+				if len(connectionState.UnmappedKeyIds()) > 0 {
+					repo, err := c.WorkerAuthRepoStorageFn()
+					if err != nil {
+						event.WriteError(cancelCtx, op, err, event.WithInfoMsg("error fetching worker auth repository for cluster connection maintenance"))
+						break
+					}
+					authorized, err := repo.FilterToAuthorizedWorkerKeyIds(cancelCtx, connectionState.UnmappedKeyIds())
+					if err != nil {
+						event.WriteError(cancelCtx, op, err, event.WithInfoMsg("couldn't get authorized workers from repo"))
+						break
+					}
+					connectionState.DisconnectMissingKeyIds(authorized)
 				}
-				cluster.DisconnectUnauthorized(m, wKeyIds, authorized)
 			}
 
 			timer.Reset(getRandomInterval())

internal/daemon/worker/controller_connection.go

@@ -23,10 +23,12 @@ import (
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/cmd/base"
+	"github.com/hashicorp/boundary/internal/daemon/cluster"
 	"github.com/hashicorp/boundary/internal/daemon/worker/internal/metric"
 	"github.com/hashicorp/boundary/internal/errors"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/version"
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/nodeenrollment"
 	"github.com/hashicorp/nodeenrollment/multihop"
@@ -84,10 +86,10 @@ func (w *Worker) StartControllerConnections() error {
 // upstreamDialerFunc dials an upstream server. extraAlpnProtos can be provided
 // to kms and pki connections and are used for identifying, on the server side,
 // the intended purpose of the connection.  upstreamDialerFunc takes an optional
-// state struct which, if the connection is made using the node enrolment
-// library, sends the state to the server and can be retrieved from the
-// resulting protocol.Conn.
-func (w *Worker) upstreamDialerFunc(state *structpb.Struct, extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) {
+// stateProvidingFunction which, if the connection is made using the nodeenrollment
+// library, sends the state provided by it to the server and can be retrieved
+// from the resulting *protocol.Conn.
+func (w *Worker) upstreamDialerFunc(extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) {
 	const op = "worker.(Worker).upstreamDialerFunc"
 	return func(ctx context.Context, addr string) (net.Conn, error) {
 		var conn net.Conn
@@ -96,11 +98,16 @@ func (w *Worker) upstreamDialerFunc(state *structpb.Struct, extraAlpnProtos ...s
 		case w.conf.WorkerAuthKms != nil && !w.conf.DevUsePkiForUpstream:
 			conn, err = w.v1KmsAuthDialFn(ctx, addr, extraAlpnProtos...)
 		default:
+			var st *structpb.Struct
+			st, err = w.workerConnectionInfo(addr)
+			if err != nil {
+				event.WriteError(w.baseContext, op, err)
+			}
 			conn, err = protocol.Dial(
 				ctx,
 				w.WorkerAuthStorage,
 				addr,
-				nodeenrollment.WithState(state),
+				nodeenrollment.WithState(st),
 				nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms),
 				nodeenrollment.WithExtraAlpnProtos(extraAlpnProtos),
 				// If the activation token hasn't been populated, this won't do
@@ -213,7 +220,7 @@ func (w *Worker) createClientConn(addr string) error {
 		grpc.WithUnaryInterceptor(metric.InstrumentClusterClient()),
 		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(math.MaxInt32)),
 		grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(math.MaxInt32)),
-		grpc.WithContextDialer(w.upstreamDialerFunc(nil)),
+		grpc.WithContextDialer(w.upstreamDialerFunc()),
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithDefaultServiceConfig(defServiceConfig),
 		// Don't have the resolver reach out for a service config from the
@@ -347,3 +354,23 @@ func (w *Worker) workerAuthTLSConfig(extraAlpnProtos ...string) (*tls.Config, *b
 
 	return tlsConfig, info, nil
 }
+
+// workerConnectionInfo returns the worker's cluster.WorkerConnectionInfo as
+// a struct suitable to be used as the state field in a *protocol.Conn
+func (w *Worker) workerConnectionInfo(addr string) (*structpb.Struct, error) {
+	const op = "worker.(Worker).workerConnectionInfo"
+	wci := &cluster.WorkerConnectionInfo{
+		UpstreamAddress: addr,
+		BoundaryVersion: version.Get().VersionNumber(),
+	}
+	if w.LastStatusSuccess() != nil && len(w.LastStatusSuccess().GetWorkerId()) > 0 {
+		// even though we wont have the worker the first time we dial, any
+		// redial attempts should result in the worker id being populated
+		wci.WorkerId = w.LastStatusSuccess().GetWorkerId()
+	}
+	st, err := wci.AsConnectionStateStruct()
+	if err != nil {
+		return nil, errors.Wrap(w.baseContext, err, op, errors.WithMsg("getting worker state"))
+	}
+	return st, nil
+}

internal/daemon/worker/listeners.go

@@ -32,6 +32,13 @@ import (
 	"google.golang.org/grpc"
 )
 
+const (
+	// the purpose strings used to identify listeners
+	reverseGrpcListenerPurpose            = "reverse-grpc"
+	multihopProxyDataplaneListenerPurpose = "multihop-proxy-dataplane"
+	grpcListenerPurpose                   = "grpc"
+)
+
 // the function that handles a secondary connection over a provided listener
 var handleSecondaryConnection = closeListeners
 
@@ -162,7 +169,7 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 	// This handles connections coming in that are authenticated via
 	// nodeenrollment but not with any extra purpose; these are normal PKI
 	// worker connections
-	nodeeAuthListener, err := w.workerAuthSplitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto)
+	nodeeAuthListener, err := w.workerAuthSplitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto, nodee.WithNativeConns(true))
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating worker split listener: %w", err)
 	}
@@ -183,7 +190,7 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 
 	// This wraps the reverse grpc pki worker connections with a listener which
 	// adds the worker key id of the connections to the worker's pkiConnManager.
-	revPkiWorkerTrackingListener, err := cluster.NewTrackingListener(w.baseContext, reverseGrpcListener, w.pkiConnManager)
+	revPkiWorkerTrackingListener, err := cluster.NewTrackingListener(w.baseContext, reverseGrpcListener, w.pkiConnManager, sourcePurpose(reverseGrpcListenerPurpose))
 	if err != nil {
 		return nil, fmt.Errorf("%s: error creating reverse grpc pki worker tracking listener: %w", op, err)
 	}
@@ -197,7 +204,7 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 
 	// This wraps the web socket proxying worker connections with a listener which
 	// adds the worker key id of the connections to the worker's pkiConnManager.
-	dataPlaneProxyTrackingListener, err := cluster.NewTrackingListener(w.baseContext, dataPlaneProxyListener, w.pkiConnManager)
+	dataPlaneProxyTrackingListener, err := cluster.NewTrackingListener(w.baseContext, dataPlaneProxyListener, w.pkiConnManager, sourcePurpose(multihopProxyDataplaneListenerPurpose))
 	if err != nil {
 		return nil, fmt.Errorf("%s: error creating web socket proxying tracking listener: %w", op, err)
 	}
@@ -230,17 +237,17 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 
 	// This wraps the normal pki worker connections with a listener which adds
 	// the worker key id of the  connections to the worker's pkiConnManager.
-	pkiWorkerTrackingListener, err := cluster.NewTrackingListener(cancelCtx, eventingListener, w.pkiConnManager)
+	pkiWorkerTrackingListener, err := cluster.NewTrackingListener(cancelCtx, eventingListener, w.pkiConnManager, sourcePurpose(grpcListenerPurpose))
 	if err != nil {
 		return nil, fmt.Errorf("%s: error creating pki worker tracking listener: %w", op, err)
 	}
 
 	return func() {
-		handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, "reverse-grpc"),
-			metric.InstrumentWorkerClusterTrackingListener(dataPlaneProxyTrackingListener, "multihop-proxy-dataplane"), w.downstreamReceiver)
+		handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, reverseGrpcListenerPurpose),
+			metric.InstrumentWorkerClusterTrackingListener(dataPlaneProxyTrackingListener, multihopProxyDataplaneListenerPurpose), w.downstreamReceiver)
 		go w.workerAuthSplitListener.Start()
 		go httpServer.Serve(proxyListener)
-		go ln.GrpcServer.Serve(metric.InstrumentWorkerClusterTrackingListener(pkiWorkerTrackingListener, "grpc"))
+		go ln.GrpcServer.Serve(metric.InstrumentWorkerClusterTrackingListener(pkiWorkerTrackingListener, grpcListenerPurpose))
 	}, nil
 }
 
@@ -321,3 +328,7 @@ func listenerCloseErrorCheck(lnType string, err error) error {
 
 	return err
 }
+
+func sourcePurpose(purpose string) string {
+	return fmt.Sprintf("worker %s", purpose)
+}

internal/daemon/worker/status.go

@@ -10,7 +10,6 @@ import (
 	"math/rand"
 	"time"
 
-	"github.com/hashicorp/boundary/internal/daemon/cluster"
 	"github.com/hashicorp/boundary/internal/daemon/worker/common"
 	"github.com/hashicorp/boundary/internal/daemon/worker/session"
 	pb "github.com/hashicorp/boundary/internal/gen/controller/servers"
@@ -172,7 +171,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 			event.WithInfoMsg("error making status request to controller"))
 	}
 	versionInfo := version.Get()
-	connectedWorkerKeyIds := w.pkiConnManager.Connected()
+	connectionState := w.pkiConnManager.Connected()
 	result, err := client.Status(statusCtx, &pbs.StatusRequest{
 		Jobs: activeJobs,
 		WorkerStatus: &pb.ServerWorkerStatus{
@@ -184,7 +183,8 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 			ReleaseVersion:   versionInfo.FullVersionNumber(false),
 			OperationalState: w.operationalState.Load().(server.OperationalState).String(),
 		},
-		ConnectedWorkerKeyIdentifiers: connectedWorkerKeyIds,
+		ConnectedWorkerKeyIdentifiers: connectionState.UnmappedKeyIds(),
+		ConnectedWorkerPublicIds:      connectionState.WorkerIds(),
 		UpdateTags:                    w.updateTags.Load(),
 	})
 	if err != nil {
@@ -251,7 +251,8 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 	w.updateTags.Store(false)
 
 	if authorized := result.GetAuthorizedWorkers(); authorized != nil {
-		cluster.DisconnectUnauthorized(w.pkiConnManager, connectedWorkerKeyIds, authorized.GetWorkerKeyIdentifiers())
+		connectionState.DisconnectMissingWorkers(authorized.GetWorkerPublicIds())
+		connectionState.DisconnectMissingKeyIds(authorized.GetWorkerKeyIdentifiers())
 	}
 	var addrs []string
 	// This may be empty if we are in a multiple hop scenario

internal/gen/controller/servers/services/server_coordination_service.pb.go

@@ -610,6 +610,11 @@ type StatusRequest struct {
 	// The worker key identifiers presented by all downstreams connected to this
 	// worker
 	ConnectedWorkerKeyIdentifiers []string `protobuf:"bytes,50,rep,name=connected_worker_key_identifiers,json=connectedWorkerKeyIdentifiers,proto3" json:"connected_worker_key_identifiers,omitempty"`
+	// The worker public ids of all downstreams connected to this worker if known.
+	// While there may be workers with key ids in the connected_worker_key_identifiers
+	// list and their public ids in this list, once the requesting worker is aware
+	// of the association, it should only populate this field.
+	ConnectedWorkerPublicIds []string `protobuf:"bytes,55,rep,name=connected_worker_public_ids,json=connectedWorkerPublicIds,proto3" json:"connected_worker_public_ids,omitempty"`
 }
 
 func (x *StatusRequest) Reset() {
@@ -672,6 +677,13 @@ func (x *StatusRequest) GetConnectedWorkerKeyIdentifiers() []string {
 	return nil
 }
 
+func (x *StatusRequest) GetConnectedWorkerPublicIds() []string {
+	if x != nil {
+		return x.ConnectedWorkerPublicIds
+	}
+	return nil
+}
+
 type JobChangeRequest struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -732,7 +744,11 @@ type AuthorizedWorkerList struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
+	// the key id of authorized workers
 	WorkerKeyIdentifiers []string `protobuf:"bytes,1,rep,name=worker_key_identifiers,json=workerKeyIdentifiers,proto3" json:"worker_key_identifiers,omitempty"`
+	// the public id of workers that were reported as being connected and which
+	// are known by the controller.
+	WorkerPublicIds []string `protobuf:"bytes,2,rep,name=worker_public_ids,json=workerPublicIds,proto3" json:"worker_public_ids,omitempty"`
 }
 
 func (x *AuthorizedWorkerList) Reset() {
@@ -774,6 +790,13 @@ func (x *AuthorizedWorkerList) GetWorkerKeyIdentifiers() []string {
 	return nil
 }
 
+func (x *AuthorizedWorkerList) GetWorkerPublicIds() []string {
+	if x != nil {
+		return x.WorkerPublicIds
+	}
+	return nil
+}
+
 type StatusResponse struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -1061,7 +1084,7 @@ var file_controller_servers_services_v1_server_coordination_service_proto_rawDes
 	0x0a, 0x10, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49,
 	0x45, 0x44, 0x10, 0x00, 0x12, 0x13, 0x0a, 0x0f, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x43, 0x4f, 0x4e,
 	0x54, 0x52, 0x4f, 0x4c, 0x4c, 0x45, 0x52, 0x10, 0x01, 0x12, 0x0f, 0x0a, 0x0b, 0x54, 0x59, 0x50,
-	0x45, 0x5f, 0x57, 0x4f, 0x52, 0x4b, 0x45, 0x52, 0x10, 0x02, 0x22, 0x96, 0x02, 0x0a, 0x0d, 0x53,
+	0x45, 0x5f, 0x57, 0x4f, 0x52, 0x4b, 0x45, 0x52, 0x10, 0x02, 0x22, 0xd5, 0x02, 0x0a, 0x0d, 0x53,
 	0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3d, 0x0a, 0x04,
 	0x6a, 0x6f, 0x62, 0x73, 0x18, 0x14, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x29, 0x2e, 0x63, 0x6f, 0x6e,
 	0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e,
@@ -1078,106 +1101,112 @@ var file_controller_servers_services_v1_server_coordination_service_proto_rawDes
 	0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x66, 0x69, 0x65, 0x72, 0x73,
 	0x18, 0x32, 0x20, 0x03, 0x28, 0x09, 0x52, 0x1d, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x65,
 	0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4b, 0x65, 0x79, 0x49, 0x64, 0x65, 0x6e, 0x74, 0x69,
-	0x66, 0x69, 0x65, 0x72, 0x73, 0x4a, 0x04, 0x08, 0x0a, 0x10, 0x0b, 0x52, 0x06, 0x77, 0x6f, 0x72,
-	0x6b, 0x65, 0x72, 0x22, 0x98, 0x01, 0x0a, 0x10, 0x4a, 0x6f, 0x62, 0x43, 0x68, 0x61, 0x6e, 0x67,
-	0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x35, 0x0a, 0x03, 0x6a, 0x6f, 0x62, 0x18,
-	0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c,
-	0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69,
-	0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4a, 0x6f, 0x62, 0x52, 0x03, 0x6a, 0x6f, 0x62, 0x12,
-	0x4d, 0x0a, 0x0c, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c,
-	0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69,
-	0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50,
-	0x45, 0x52, 0x0b, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x22, 0x4c,
-	0x0a, 0x14, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b,
-	0x65, 0x72, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x34, 0x0a, 0x16, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72,
-	0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x66, 0x69, 0x65, 0x72, 0x73,
-	0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x14, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4b, 0x65,
-	0x79, 0x49, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x66, 0x69, 0x65, 0x72, 0x73, 0x22, 0xdf, 0x02, 0x0a,
-	0x0e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12,
-	0x55, 0x0a, 0x0d, 0x6a, 0x6f, 0x62, 0x73, 0x5f, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x73,
-	0x18, 0x14, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x30, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c,
-	0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76,
-	0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4a, 0x6f, 0x62, 0x43, 0x68, 0x61, 0x6e, 0x67,
-	0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x0c, 0x6a, 0x6f, 0x62, 0x73, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x73, 0x12, 0x61, 0x0a, 0x14, 0x63, 0x61, 0x6c, 0x63, 0x75, 0x6c,
-	0x61, 0x74, 0x65, 0x64, 0x5f, 0x75, 0x70, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x73, 0x18, 0x1e,
-	0x20, 0x03, 0x28, 0x0b, 0x32, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65,
+	0x66, 0x69, 0x65, 0x72, 0x73, 0x12, 0x3d, 0x0a, 0x1b, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x65, 0x64, 0x5f, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63,
+	0x5f, 0x69, 0x64, 0x73, 0x18, 0x37, 0x20, 0x03, 0x28, 0x09, 0x52, 0x18, 0x63, 0x6f, 0x6e, 0x6e,
+	0x65, 0x63, 0x74, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x50, 0x75, 0x62, 0x6c, 0x69,
+	0x63, 0x49, 0x64, 0x73, 0x4a, 0x04, 0x08, 0x0a, 0x10, 0x0b, 0x52, 0x06, 0x77, 0x6f, 0x72, 0x6b,
+	0x65, 0x72, 0x22, 0x98, 0x01, 0x0a, 0x10, 0x4a, 0x6f, 0x62, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x35, 0x0a, 0x03, 0x6a, 0x6f, 0x62, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65,
 	0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63,
-	0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x55, 0x70, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x65,
-	0x72, 0x76, 0x65, 0x72, 0x52, 0x13, 0x63, 0x61, 0x6c, 0x63, 0x75, 0x6c, 0x61, 0x74, 0x65, 0x64,
-	0x55, 0x70, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x77, 0x6f, 0x72,
-	0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x77, 0x6f,
-	0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x63, 0x0a, 0x12, 0x61, 0x75, 0x74, 0x68, 0x6f, 0x72,
-	0x69, 0x7a, 0x65, 0x64, 0x5f, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x18, 0x32, 0x20, 0x01,
-	0x28, 0x0b, 0x32, 0x34, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e,
-	0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73,
-	0x2e, 0x76, 0x31, 0x2e, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x64, 0x57, 0x6f,
-	0x72, 0x6b, 0x65, 0x72, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x11, 0x61, 0x75, 0x74, 0x68, 0x6f, 0x72,
-	0x69, 0x7a, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x4a, 0x04, 0x08, 0x0a, 0x10,
-	0x0b, 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x73, 0x22, 0x36,
-	0x0a, 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x0e, 0x0a, 0x02,
-	0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x18, 0x0a, 0x07,
-	0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61,
-	0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x22, 0x18, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63,
-	0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x22, 0x5f, 0x0a, 0x17, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b,
-	0x65, 0x72, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x44, 0x0a, 0x07, 0x77,
-	0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63,
-	0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
-	0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x57, 0x6f,
-	0x72, 0x6b, 0x65, 0x72, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x07, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72,
-	0x73, 0x2a, 0x92, 0x01, 0x0a, 0x10, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e,
-	0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x12, 0x20, 0x0a, 0x1c, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43,
-	0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45,
-	0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1f, 0x0a, 0x1b, 0x43, 0x4f, 0x4e, 0x4e,
-	0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x41, 0x55, 0x54,
-	0x48, 0x4f, 0x52, 0x49, 0x5a, 0x45, 0x44, 0x10, 0x01, 0x12, 0x1e, 0x0a, 0x1a, 0x43, 0x4f, 0x4e,
-	0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x43, 0x4f,
-	0x4e, 0x4e, 0x45, 0x43, 0x54, 0x45, 0x44, 0x10, 0x02, 0x12, 0x1b, 0x0a, 0x17, 0x43, 0x4f, 0x4e,
-	0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x43, 0x4c,
-	0x4f, 0x53, 0x45, 0x44, 0x10, 0x03, 0x2a, 0x9e, 0x01, 0x0a, 0x0d, 0x53, 0x45, 0x53, 0x53, 0x49,
-	0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x12, 0x1d, 0x0a, 0x19, 0x53, 0x45, 0x53, 0x53,
-	0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43,
-	0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x53, 0x45, 0x53, 0x53, 0x49,
-	0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x50, 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47,
-	0x10, 0x01, 0x12, 0x18, 0x0a, 0x14, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41,
-	0x54, 0x55, 0x53, 0x5f, 0x41, 0x43, 0x54, 0x49, 0x56, 0x45, 0x10, 0x02, 0x12, 0x1b, 0x0a, 0x17,
-	0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x43, 0x41,
-	0x4e, 0x43, 0x45, 0x4c, 0x49, 0x4e, 0x47, 0x10, 0x03, 0x12, 0x1c, 0x0a, 0x18, 0x53, 0x45, 0x53,
-	0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x54, 0x45, 0x52, 0x4d, 0x49,
-	0x4e, 0x41, 0x54, 0x45, 0x44, 0x10, 0x04, 0x2a, 0x37, 0x0a, 0x07, 0x4a, 0x4f, 0x42, 0x54, 0x59,
-	0x50, 0x45, 0x12, 0x17, 0x0a, 0x13, 0x4a, 0x4f, 0x42, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e,
-	0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x13, 0x0a, 0x0f, 0x4a,
-	0x4f, 0x42, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x10, 0x01,
-	0x2a, 0x45, 0x0a, 0x0a, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x12, 0x1a,
-	0x0a, 0x16, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x53,
-	0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1b, 0x0a, 0x17, 0x43, 0x48,
-	0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x50, 0x44, 0x41, 0x54, 0x45, 0x5f,
-	0x53, 0x54, 0x41, 0x54, 0x45, 0x10, 0x01, 0x32, 0x8d, 0x02, 0x0a, 0x19, 0x53, 0x65, 0x72, 0x76,
-	0x65, 0x72, 0x43, 0x6f, 0x6f, 0x72, 0x64, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x65,
-	0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x69, 0x0a, 0x06, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12,
-	0x2d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72,
+	0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4a, 0x6f, 0x62, 0x52, 0x03, 0x6a, 0x6f, 0x62, 0x12, 0x4d,
+	0x0a, 0x0c, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x02,
+	0x20, 0x01, 0x28, 0x0e, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65,
+	0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63,
+	0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45,
+	0x52, 0x0b, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x22, 0x78, 0x0a,
+	0x14, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65,
+	0x72, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x34, 0x0a, 0x16, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f,
+	0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x66, 0x69, 0x65, 0x72, 0x73, 0x18,
+	0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x14, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4b, 0x65, 0x79,
+	0x49, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x66, 0x69, 0x65, 0x72, 0x73, 0x12, 0x2a, 0x0a, 0x11, 0x77,
+	0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x69, 0x64, 0x73,
+	0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0f, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x50, 0x75,
+	0x62, 0x6c, 0x69, 0x63, 0x49, 0x64, 0x73, 0x22, 0xdf, 0x02, 0x0a, 0x0e, 0x53, 0x74, 0x61, 0x74,
+	0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x55, 0x0a, 0x0d, 0x6a, 0x6f,
+	0x62, 0x73, 0x5f, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x73, 0x18, 0x14, 0x20, 0x03, 0x28,
+	0x0b, 0x32, 0x30, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73,
+	0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e,
+	0x76, 0x31, 0x2e, 0x4a, 0x6f, 0x62, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x52, 0x0c, 0x6a, 0x6f, 0x62, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x73, 0x12, 0x61, 0x0a, 0x14, 0x63, 0x61, 0x6c, 0x63, 0x75, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x5f,
+	0x75, 0x70, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x73, 0x18, 0x1e, 0x20, 0x03, 0x28, 0x0b, 0x32,
+	0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72,
 	0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31,
-	0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e,
-	0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76,
-	0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e,
-	0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00,
-	0x12, 0x84, 0x01, 0x0a, 0x0f, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72,
-	0x6b, 0x65, 0x72, 0x73, 0x12, 0x36, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65,
-	0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63,
-	0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f,
-	0x72, 0x6b, 0x65, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x37, 0x2e, 0x63,
-	0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
-	0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4c, 0x69,
-	0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x52, 0x65, 0x73,
-	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x51, 0x5a, 0x4f, 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, 0x67, 0x65, 0x6e, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72,
-	0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65,
-	0x73, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x33,
+	0x2e, 0x55, 0x70, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x52,
+	0x13, 0x63, 0x61, 0x6c, 0x63, 0x75, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x55, 0x70, 0x73, 0x74, 0x72,
+	0x65, 0x61, 0x6d, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69,
+	0x64, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49,
+	0x64, 0x12, 0x63, 0x0a, 0x12, 0x61, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x64, 0x5f,
+	0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x18, 0x32, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x34, 0x2e,
+	0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65,
+	0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x41,
+	0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4c,
+	0x69, 0x73, 0x74, 0x52, 0x11, 0x61, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x64, 0x57,
+	0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x4a, 0x04, 0x08, 0x0a, 0x10, 0x0b, 0x52, 0x0b, 0x63, 0x6f,
+	0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x73, 0x22, 0x36, 0x0a, 0x0a, 0x57, 0x6f, 0x72,
+	0x6b, 0x65, 0x72, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65,
+	0x73, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73,
+	0x73, 0x22, 0x18, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72,
+	0x6b, 0x65, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x5f, 0x0a, 0x17, 0x4c,
+	0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x52, 0x65,
+	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x44, 0x0a, 0x07, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72,
+	0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f,
+	0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72,
+	0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49,
+	0x6e, 0x66, 0x6f, 0x52, 0x07, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x2a, 0x92, 0x01, 0x0a,
+	0x10, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55,
+	0x53, 0x12, 0x20, 0x0a, 0x1c, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53,
+	0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45,
+	0x44, 0x10, 0x00, 0x12, 0x1f, 0x0a, 0x1b, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f,
+	0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x41, 0x55, 0x54, 0x48, 0x4f, 0x52, 0x49, 0x5a,
+	0x45, 0x44, 0x10, 0x01, 0x12, 0x1e, 0x0a, 0x1a, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49,
+	0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54,
+	0x45, 0x44, 0x10, 0x02, 0x12, 0x1b, 0x0a, 0x17, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49,
+	0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x43, 0x4c, 0x4f, 0x53, 0x45, 0x44, 0x10,
+	0x03, 0x2a, 0x9e, 0x01, 0x0a, 0x0d, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41,
+	0x54, 0x55, 0x53, 0x12, 0x1d, 0x0a, 0x19, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54,
+	0x41, 0x54, 0x55, 0x53, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44,
+	0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41,
+	0x54, 0x55, 0x53, 0x5f, 0x50, 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47, 0x10, 0x01, 0x12, 0x18, 0x0a,
+	0x14, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x41,
+	0x43, 0x54, 0x49, 0x56, 0x45, 0x10, 0x02, 0x12, 0x1b, 0x0a, 0x17, 0x53, 0x45, 0x53, 0x53, 0x49,
+	0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x43, 0x41, 0x4e, 0x43, 0x45, 0x4c, 0x49,
+	0x4e, 0x47, 0x10, 0x03, 0x12, 0x1c, 0x0a, 0x18, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53,
+	0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x54, 0x45, 0x52, 0x4d, 0x49, 0x4e, 0x41, 0x54, 0x45, 0x44,
+	0x10, 0x04, 0x2a, 0x37, 0x0a, 0x07, 0x4a, 0x4f, 0x42, 0x54, 0x59, 0x50, 0x45, 0x12, 0x17, 0x0a,
+	0x13, 0x4a, 0x4f, 0x42, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49,
+	0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x13, 0x0a, 0x0f, 0x4a, 0x4f, 0x42, 0x54, 0x59, 0x50,
+	0x45, 0x5f, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x10, 0x01, 0x2a, 0x45, 0x0a, 0x0a, 0x43,
+	0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x12, 0x1a, 0x0a, 0x16, 0x43, 0x48, 0x41,
+	0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46,
+	0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1b, 0x0a, 0x17, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54,
+	0x59, 0x50, 0x45, 0x5f, 0x55, 0x50, 0x44, 0x41, 0x54, 0x45, 0x5f, 0x53, 0x54, 0x41, 0x54, 0x45,
+	0x10, 0x01, 0x32, 0x8d, 0x02, 0x0a, 0x19, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x43, 0x6f, 0x6f,
+	0x72, 0x64, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65,
+	0x12, 0x69, 0x0a, 0x06, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x2d, 0x2e, 0x63, 0x6f, 0x6e,
+	0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e,
+	0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x61, 0x74,
+	0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74,
+	0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73,
+	0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75,
+	0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x84, 0x01, 0x0a, 0x0f,
+	0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x12,
+	0x36, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72,
+	0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31,
+	0x2e, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x37, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f,
+	0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72,
+	0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70,
+	0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x00, 0x42, 0x51, 0x5a, 0x4f, 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, 0x67, 0x65, 0x6e,
+	0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x65, 0x72, 0x76,
+	0x65, 0x72, 0x73, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x3b, 0x73, 0x65, 0x72,
+	0x76, 0x69, 0x63, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (

internal/proto/controller/servers/services/v1/server_coordination_service.proto

@@ -102,6 +102,11 @@ message StatusRequest {
   // The worker key identifiers presented by all downstreams connected to this
   // worker
   repeated string connected_worker_key_identifiers = 50;
+  // The worker public ids of all downstreams connected to this worker if known.
+  // While there may be workers with key ids in the connected_worker_key_identifiers
+  // list and their public ids in this list, once the requesting worker is aware
+  // of the association, it should only populate this field.
+  repeated string connected_worker_public_ids = 55;
 }
 
 enum CHANGETYPE {
@@ -117,7 +122,11 @@ message JobChangeRequest {
 }
 
 message AuthorizedWorkerList {
+  // the key id of authorized workers
   repeated string worker_key_identifiers = 1;
+  // the public id of workers that were reported as being connected and which
+  // are known by the controller.
+  repeated string worker_public_ids = 2;
 }
 
 message StatusResponse {

internal/server/repository_worker.go

@@ -166,7 +166,9 @@ func lookupWorker(ctx context.Context, reader db.Reader, id string) (*Worker, er
 // 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.
+// 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.
 // Also supports: WithWorkerType, WithActiveWorkers
 func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ...Option) ([]*Worker, error) {
 	const op = "server.(Repository).ListWorkers"
@@ -205,6 +207,11 @@ func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ...
 		whereArgs = append(whereArgs, ActiveOperationalState.String())
 	}
 
+	if len(opts.withWorkerPool) > 0 {
+		where = append(where, "public_id in (?)")
+		whereArgs = append(whereArgs, opts.withWorkerPool)
+	}
+
 	limit := r.defaultLimit
 	if opts.withLimit != 0 {
 		// non-zero signals an override of the default limit for the repo.

internal/server/repository_worker_test.go

@@ -648,6 +648,62 @@ func TestListWorkers(t *testing.T) {
 	}
 }
 
+func TestListWorkers_WithWorkerPool(t *testing.T) {
+	t.Parallel()
+	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(rw, rw, kmsCache)
+	require.NoError(err)
+	ctx := context.Background()
+
+	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()
 	require := require.New(t)