Add cleanup of dead connections no longer reported by a worker (#1220)

CHANGELOG.md

@@ -4,6 +4,15 @@ Canonical reference for changes, improvements, and bugfixes for Boundary.
 
 ## Next
 
+### Bug Fixes
+
+* sessions: Clean up connections that are dangling after a worker dies (is
+  restarted, powered off, etc.) This fixes some cases where a session never goes
+  to `terminated` state because connections are not properly marked closed.
+  ([Issue 1](https://github.com/hashicorp/boundary/issues/894), [Issue
+  2](https://github.com/hashicorp/boundary/issues/1055),
+  [PR](https://github.com/hashicorp/boundary/pull/1220))
+
 ## 0.2.1 (2021/05/05)
 
 ### Deprecations/Changes

internal/cmd/commands/dev/dev.go

@@ -62,6 +62,8 @@ type Command struct {
 	flagControllerAPIListenAddr      string
 	flagControllerClusterListenAddr  string
 	flagControllerPublicClusterAddr  string
+	flagControllerOnly               bool
+	flagWorkerAuthKey                string
 	flagWorkerProxyListenAddr        string
 	flagWorkerPublicAddr             string
 	flagPassthroughDirectory         string
@@ -202,6 +204,12 @@ func (c *Command) Flags() *base.FlagSets {
 		Usage:  "Public address at which the controller is reachable for cluster tasks (like worker connections).",
 	})
 
+	f.BoolVar(&base.BoolVar{
+		Name:   "controller-only",
+		Target: &c.flagControllerOnly,
+		Usage:  "If set, only a dev controller will be started instead of both a dev controller and dev worker",
+	})
+
 	f.StringVar(&base.StringVar{
 		Name:   "proxy-listen-address",
 		Target: &c.flagWorkerProxyListenAddr,
@@ -216,6 +224,13 @@ func (c *Command) Flags() *base.FlagSets {
 		Usage:  "Public address at which the worker is reachable for session proxying.",
 	})
 
+	f.StringVar(&base.StringVar{
+		Name:   "worker-auth-key",
+		Target: &c.flagWorkerAuthKey,
+		EnvVar: "BOUNDARY_DEV_WORKER_AUTH_KEY",
+		Usage:  "If set, a valid base64-encoded AES key to be used for worker-auth purposes",
+	})
+
 	f.BoolVar(&base.BoolVar{
 		Name:   "disable-database-destruction",
 		Target: &c.flagDisableDatabaseDestruction,
@@ -276,11 +291,24 @@ func (c *Command) Run(args []string) int {
 		return base.CommandUserError
 	}
 
-	c.Config, err = config.DevCombined()
+	switch c.flagControllerOnly {
+	case true:
+		c.Config, err = config.DevController()
+	default:
+		c.Config, err = config.DevCombined()
+	}
 	if err != nil {
 		c.UI.Error(fmt.Errorf("Error creating controller dev config: %w", err).Error())
 		return base.CommandUserError
 	}
+	if c.flagWorkerAuthKey != "" {
+		c.Config.DevWorkerAuthKey = c.flagWorkerAuthKey
+		for _, kms := range c.Config.Seals {
+			if strutil.StrListContains(kms.Purpose, "worker-auth") {
+				kms.Config["key"] = c.flagWorkerAuthKey
+			}
+		}
+	}
 	if c.flagIdSuffix != "" {
 		if len(c.flagIdSuffix) != 10 {
 			c.UI.Error("Invalid ID suffix, must be exactly 10 characters")
@@ -353,7 +381,9 @@ func (c *Command) Run(args []string) int {
 		case "cluster":
 			if c.flagControllerClusterListenAddr != "" {
 				l.Address = c.flagControllerClusterListenAddr
-				c.Config.Worker.Controllers = []string{l.Address}
+				if !c.flagControllerOnly {
+					c.Config.Worker.Controllers = []string{l.Address}
+				}
 			} else {
 				l.Address = "127.0.0.1:9201"
 			}
@@ -381,8 +411,10 @@ func (c *Command) Run(args []string) int {
 		c.UI.Error(err.Error())
 		return base.CommandUserError
 	}
-	c.InfoKeys = append(c.InfoKeys, "worker public proxy addr")
+	if !c.flagControllerOnly {
-	c.Info["worker public proxy addr"] = c.Config.Worker.PublicAddr
+		c.InfoKeys = append(c.InfoKeys, "worker public proxy addr")
+		c.Info["worker public proxy addr"] = c.Config.Worker.PublicAddr
+	}
 
 	if err := c.SetupLogging(c.flagLogLevel, c.flagLogFormat, "", ""); err != nil {
 		c.UI.Error(err.Error())
@@ -496,7 +528,8 @@ func (c *Command) Run(args []string) int {
 			return base.CommandCliError
 		}
 	}
-	{
+
+	if !c.flagControllerOnly {
 		conf := &worker.Config{
 			RawConfig: c.Config,
 			Server:    c.Server,
@@ -531,8 +564,10 @@ func (c *Command) Run(args []string) int {
 		case <-c.ShutdownCh:
 			c.UI.Output("==> Boundary dev environment shutdown triggered")
 
-			if err := c.worker.Shutdown(false); err != nil {
+			if !c.flagControllerOnly {
-				c.UI.Error(fmt.Errorf("Error shutting down worker: %w", err).Error())
+				if err := c.worker.Shutdown(false); err != nil {
+					c.UI.Error(fmt.Errorf("Error shutting down worker: %w", err).Error())
+				}
 			}
 
 			if err := c.controller.Shutdown(false); err != nil {

internal/db/read_writer_test.go

@@ -10,7 +10,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/hashicorp/boundary/internal/db/db_test"
 	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/errors"
@@ -23,6 +22,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 func TestDb_UpdateUnsetField(t *testing.T) {
@@ -50,7 +50,7 @@ func TestDb_UpdateUnsetField(t *testing.T) {
 
 func TestDb_Update(t *testing.T) {
 	db, _ := TestSetup(t, "postgres")
-	now := &timestamp.Timestamp{Timestamp: ptypes.TimestampNow()}
+	now := &timestamp.Timestamp{Timestamp: timestamppb.Now()}
 	publicId, err := NewPublicId("testuser")
 	require.NoError(t, err)
 	id := testId(t)
@@ -623,7 +623,7 @@ func TestDb_Create(t *testing.T) {
 		require.NoError(err)
 		user, err := db_test.NewTestUser()
 		require.NoError(err)
-		ts := &timestamp.Timestamp{Timestamp: ptypes.TimestampNow()}
+		ts := &timestamp.Timestamp{Timestamp: timestamppb.Now()}
 		user.CreateTime = ts
 		user.UpdateTime = ts
 		user.Name = "foo-" + id

internal/db/schema/migrations/postgres/0/50_session.up.sql

@@ -89,6 +89,9 @@ begin;
     ('connection limit'),
     ('canceled');
 
+-- Note: here, and in the session_connection table, we should add a trigger
+-- ensuring that if server_id goes to null, we mark connections as closed. See
+-- https://hashicorp.atlassian.net/browse/ICU-1495
   create table session (
     public_id wt_public_id primary key,
     -- the user of the session

internal/db/schema/migrations/postgres/0/51_connection.up.sql

@@ -79,6 +79,7 @@ begin;
   -- a endpoint for a session. The client initiates the connection to the worker
   -- and the worker initiates the connection to the endpoint.
   -- A session can have zero or more session connections.
+  -- Note: Updated to add server_id, server_type in 801
   create table session_connection (
     public_id wt_public_id primary key,
     session_id wt_public_id not null

internal/db/schema/migrations/postgres/8/01_connection.up.sql

@@ -0,0 +1,25 @@
+begin;
+
+alter table session_connection
+  add column server_id text;
+
+-- Note: here, and in the session table, we should add a trigger ensuring that
+-- if server_id goes to null, we mark connections as closed. See
+-- https://hashicorp.atlassian.net/browse/ICU-1495
+alter table session_connection
+  add constraint server_fkey
+    foreign key (server_id)
+    references server (private_id)
+    on delete set null
+    on update cascade;
+
+-- We now populate the connection information from existing session information
+update session_connection sc
+set
+  server_id = s.server_id
+from
+  session s
+where
+  sc.session_id = s.public_id;
+
+commit;

internal/db/schema/migrations/postgres/8/08_connection_test.go

@@ -0,0 +1,102 @@
+package migration
+
+import (
+	"database/sql"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMigration(t *testing.T) {
+	require := require.New(t)
+	const (
+		createTables = `
+		create table session_testing (
+			public_id text primary key,
+			server_id text
+		);
+		create table session_connection_testing (
+			public_id text primary key,
+			session_id text,
+			server_id text
+		);
+		
+	`
+		insertSession           = `insert into session_testing values ($1, $2)`
+		insertSessionConnection = `insert into session_connection_testing values ($1, $2)`
+
+		update = `
+		update session_connection_testing sc
+		set
+		  server_id = s.server_id
+		from
+		  session_testing s
+		where
+		  sc.session_id = s.public_id;
+		`
+		selectQuery = `select session_id, server_id from session_connection_testing order by session_id`
+	)
+
+	conn, _ := db.TestSetup(t, "postgres")
+	db := conn.DB()
+	_, err := db.Exec(createTables)
+	require.NoError(err)
+
+	tests := []struct {
+		sessId       string
+		sessServerId string
+		connId       string
+	}{
+		{
+			"sess1",
+			"server1",
+			"conn1",
+		},
+		{
+			"sess2",
+			"server1",
+			"conn2",
+		},
+		{
+			"sess3",
+			"server2",
+			"conn3",
+		},
+		{
+			"sess4",
+			"server2",
+			"conn4",
+		},
+	}
+	for _, tt := range tests {
+		_, err = db.Query(insertSession, tt.sessId, tt.sessServerId)
+		require.NoError(err)
+		_, err := db.Query(insertSessionConnection, tt.connId, tt.sessId)
+		require.NoError(err)
+	}
+
+	// At this point server ID should be empty in connections table
+	rows, err := db.Query(selectQuery)
+	require.NoError(err)
+	var sessVal string
+	var serverVal sql.NullString
+	for rows.Next() {
+		require.NoError(rows.Scan(&sessVal, &serverVal))
+		require.False(serverVal.Valid)
+	}
+
+	_, err = db.Query(update)
+	require.NoError(err)
+
+	// Now it should match the server
+	rows, err = db.Query(selectQuery)
+	require.NoError(err)
+	var count int
+	for rows.Next() {
+		require.NoError(rows.Scan(&sessVal, &serverVal))
+		require.True(serverVal.Valid)
+		require.Equal(tests[count].sessServerId, serverVal.String)
+		count++
+	}
+}

internal/db/schema/postgres_migration.gen.go

@@ -4,7 +4,7 @@ package schema
 
 func init() {
 	migrationStates["postgres"] = migrationState{
-		binarySchemaVersion: 7003,
+		binarySchemaVersion: 8001,
 		upMigrations: map[int][]byte{
 			1: []byte(`
 create domain wt_public_id as text
@@ -2975,6 +2975,9 @@ values
     ('connection limit'),
     ('canceled');
 
+-- Note: here, and in the session_connection table, we should add a trigger
+-- ensuring that if server_id goes to null, we mark connections as closed. See
+-- https://hashicorp.atlassian.net/browse/ICU-1495
   create table session (
     public_id wt_public_id primary key,
     -- the user of the session
@@ -3452,6 +3455,7 @@ values
   -- a endpoint for a session. The client initiates the connection to the worker
   -- and the worker initiates the connection to the endpoint.
   -- A session can have zero or more session connections.
+  -- Note: Updated to add server_id, server_type in 801
   create table session_connection (
     public_id wt_public_id primary key,
     session_id wt_public_id not null
@@ -6346,6 +6350,29 @@ create table job (
 		       )
 	  )
 	  select job_plugin_id, job_name, next_scheduled_run from final;
+`),
+			8001: []byte(`
+alter table session_connection
+  add column server_id text;
+
+-- Note: here, and in the session table, we should add a trigger ensuring that
+-- if server_id goes to null, we mark connections as closed. See
+-- https://hashicorp.atlassian.net/browse/ICU-1495
+alter table session_connection
+  add constraint server_fkey
+    foreign key (server_id)
+    references server (private_id)
+    on delete set null
+    on update cascade;
+
+-- We now populate the connection information from existing session information
+update session_connection sc
+set
+  server_id = s.server_id
+from
+  session s
+where
+  sc.session_id = s.public_id;
 `),
 		},
 	}

internal/db/timestamp/scanners.go

@@ -3,21 +3,16 @@ package timestamp
 import (
 	"database/sql/driver"
 	"errors"
-	"fmt"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 // Scan implements sql.Scanner for protobuf Timestamp.
 func (ts *Timestamp) Scan(value interface{}) error {
 	switch t := value.(type) {
 	case time.Time:
-		var err error
+		ts.Timestamp = timestamppb.New(t) // google proto version
-		ts.Timestamp, err = ptypes.TimestampProto(t) // google proto version
-		if err != nil {
-			return fmt.Errorf("error converting the timestamp: %w", err)
-		}
 	default:
 		return errors.New("Not a protobuf Timestamp")
 	}
@@ -29,5 +24,5 @@ func (ts *Timestamp) Value() (driver.Value, error) {
 	if ts == nil {
 		return nil, nil
 	}
-	return ptypes.Timestamp(ts.Timestamp)
+	return ts.Timestamp.AsTime(), nil
 }

internal/db/timestamp/scanners_test.go

@@ -7,47 +7,46 @@ import (
 
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func Test_TimestampValue(t *testing.T) {
 	t.Parallel()
-	assert := assert.New(t)
 	t.Run("valid", func(t *testing.T) {
 		ts := Timestamp{Timestamp: &timestamp.Timestamp{Seconds: 0, Nanos: 0}}
 		v, err := ts.Value()
-		assert.Nil(err)
+		require.Nil(t, err)
-		assert.Equal(v, utcDate(1970, 1, 1))
+		assert.Equal(t, v, utcDate(1970, 1, 1))
 	})
 	t.Run("valid nil ts", func(t *testing.T) {
 		var ts *Timestamp
 		v, err := ts.Value()
-		assert.Nil(err)
+		require.Nil(t, err)
-		assert.Equal(v, nil)
+		assert.Equal(t, v, nil)
 	})
 	t.Run("invalid ts", func(t *testing.T) {
 		ts := Timestamp{Timestamp: &timestamp.Timestamp{Seconds: maxValidSeconds, Nanos: 0}}
 		v, err := ts.Value()
-		assert.True(err != nil)
+		require.NoError(t, err)
-		assert.Equal("timestamp: seconds:253402300800 after 10000-01-01", err.Error())
+		assert.Equal(t, v, utcDate(10000, 1, 1))
-		assert.Equal(v, utcDate(10000, 1, 1))
 	})
 }
 
 func Test_TimestampScan(t *testing.T) {
 	t.Parallel()
-	assert := assert.New(t)
+	assert, require := assert.New(t), require.New(t)
 	t.Run("valid", func(t *testing.T) {
 		v := time.Unix(0, 0)
 		ts := Timestamp{}
 		err := ts.Scan(v)
-		assert.Nil(err)
+		require.Nil(err)
 		assert.True(reflect.DeepEqual(ts.Timestamp, &timestamp.Timestamp{Seconds: 0, Nanos: 0}))
 	})
 	t.Run("valid default time", func(t *testing.T) {
 		var v time.Time
 		ts := Timestamp{}
 		err := ts.Scan(v)
-		assert.Nil(err)
+		require.Nil(err)
 		assert.True(reflect.DeepEqual(ts.Timestamp, &timestamp.Timestamp{Seconds: -62135596800, Nanos: 0}))
 	})
 	t.Run("invalid type", func(t *testing.T) {
@@ -61,8 +60,7 @@ func Test_TimestampScan(t *testing.T) {
 		v := time.Unix(maxValidSeconds, 0)
 		ts := Timestamp{}
 		err := ts.Scan(v)
-		assert.True(err != nil)
+		require.Nil(err)
-		assert.Equal("error converting the timestamp: timestamp: seconds:253402300800 after 10000-01-01", err.Error())
 	})
 }
 

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

@@ -444,6 +444,7 @@ type AuthorizeConnectionRequest struct {
 	unknownFields protoimpl.UnknownFields
 
 	SessionId string `protobuf:"bytes,10,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"`
+	WorkerId  string `protobuf:"bytes,20,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"`
 }
 
 func (x *AuthorizeConnectionRequest) Reset() {
@@ -485,6 +486,13 @@ func (x *AuthorizeConnectionRequest) GetSessionId() string {
 	return ""
 }
 
+func (x *AuthorizeConnectionRequest) GetWorkerId() string {
+	if x != nil {
+		return x.WorkerId
+	}
+	return ""
+}
+
 type AuthorizeConnectionResponse struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -990,138 +998,140 @@ var file_controller_servers_services_v1_session_service_proto_rawDesc = []byte{
 	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, 0x45, 0x53, 0x53, 0x49,
 	0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73,
-	0x22, 0x3b, 0x0a, 0x1a, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x43, 0x6f, 0x6e,
+	0x22, 0x58, 0x0a, 0x1a, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x43, 0x6f, 0x6e,
 	0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1d,
 	0x0a, 0x0a, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x09, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x22, 0xb7, 0x01,
+	0x28, 0x09, 0x52, 0x09, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x1b, 0x0a,
-	0x0a, 0x1b, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65,
+	0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x23, 0x0a,
+	0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x22, 0xb7, 0x01, 0x0a, 0x1b, 0x41,
-	0x0d, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x0a,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e,
-	0x49, 0x64, 0x12, 0x48, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x14, 0x20, 0x01,
-	0x28, 0x0e, 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, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54,
-	0x41, 0x54, 0x55, 0x53, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x29, 0x0a, 0x10,
-	0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x5f, 0x6c, 0x65, 0x66, 0x74,
-	0x18, 0x1e, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69,
-	0x6f, 0x6e, 0x73, 0x4c, 0x65, 0x66, 0x74, 0x22, 0x87, 0x02, 0x0a, 0x18, 0x43, 0x6f, 0x6e, 0x6e,
-	0x65, 0x63, 0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69,
-	0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e,
-	0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x2c, 0x0a, 0x12, 0x63, 0x6c, 0x69,
-	0x65, 0x6e, 0x74, 0x5f, 0x74, 0x63, 0x70, 0x5f, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18,
-	0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x54, 0x63, 0x70,
-	0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x26, 0x0a, 0x0f, 0x63, 0x6c, 0x69, 0x65, 0x6e,
-	0x74, 0x5f, 0x74, 0x63, 0x70, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x0d,
-	0x52, 0x0d, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x54, 0x63, 0x70, 0x50, 0x6f, 0x72, 0x74, 0x12,
-	0x30, 0x0a, 0x14, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, 0x74, 0x63, 0x70, 0x5f,
-	0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x12, 0x65,
-	0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x54, 0x63, 0x70, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73,
-	0x73, 0x12, 0x2a, 0x0a, 0x11, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, 0x74, 0x63,
-	0x70, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x32, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0f, 0x65, 0x6e,
-	0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x54, 0x63, 0x70, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x12, 0x0a,
-	0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x3c, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70,
-	0x65, 0x22, 0x65, 0x0a, 0x19, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x43, 0x6f, 0x6e, 0x6e,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x48,
-	0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0e, 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,
-	0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53,
-	0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x93, 0x01, 0x0a, 0x1a, 0x43, 0x6c, 0x6f,
-	0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75,
-	0x65, 0x73, 0x74, 0x44, 0x61, 0x74, 0x61, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x6e, 0x65,
-	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c,
-	0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x19, 0x0a, 0x08,
-	0x62, 0x79, 0x74, 0x65, 0x73, 0x5f, 0x75, 0x70, 0x18, 0x14, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07,
-	0x62, 0x79, 0x74, 0x65, 0x73, 0x55, 0x70, 0x12, 0x1d, 0x0a, 0x0a, 0x62, 0x79, 0x74, 0x65, 0x73,
-	0x5f, 0x64, 0x6f, 0x77, 0x6e, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x62, 0x79, 0x74,
-	0x65, 0x73, 0x44, 0x6f, 0x77, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e,
-	0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e, 0x22, 0x82,
-	0x01, 0x0a, 0x16, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69,
-	0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x68, 0x0a, 0x12, 0x63, 0x6c, 0x6f,
-	0x73, 0x65, 0x5f, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x5f, 0x64, 0x61, 0x74, 0x61, 0x18,
-	0x0a, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x3a, 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, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x44, 0x61, 0x74,
-	0x61, 0x52, 0x10, 0x63, 0x6c, 0x6f, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x44,
-	0x61, 0x74, 0x61, 0x22, 0x8c, 0x01, 0x0a, 0x1b, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e,
-	0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x44,
-	0x61, 0x74, 0x61, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f,
-	0x6e, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x6e,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x48, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74,
-	0x75, 0x73, 0x18, 0x14, 0x20, 0x01, 0x28, 0x0e, 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, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43,
-	0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74,
-	0x75, 0x73, 0x22, 0x86, 0x01, 0x0a, 0x17, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x6b,
-	0x0a, 0x13, 0x63, 0x6c, 0x6f, 0x73, 0x65, 0x5f, 0x72, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x5f, 0x64, 0x61, 0x74, 0x61, 0x18, 0x0a, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x3b, 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, 0x6c, 0x6f,
-	0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70,
-	0x6f, 0x6e, 0x73, 0x65, 0x44, 0x61, 0x74, 0x61, 0x52, 0x11, 0x63, 0x6c, 0x6f, 0x73, 0x65, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x44, 0x61, 0x74, 0x61, 0x32, 0xbe, 0x06, 0x0a, 0x0e,
-	0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x7e,
-	0x0a, 0x0d, 0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12,
-	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, 0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x35, 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, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x53, 0x65, 0x73,
-	0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x84,
-	0x01, 0x0a, 0x0f, 0x41, 0x63, 0x74, 0x69, 0x76, 0x61, 0x74, 0x65, 0x53, 0x65, 0x73, 0x73, 0x69,
-	0x6f, 0x6e, 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, 0x41, 0x63, 0x74, 0x69, 0x76, 0x61, 0x74, 0x65, 0x53, 0x65, 0x73, 0x73,
-	0x69, 0x6f, 0x6e, 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, 0x41, 0x63, 0x74, 0x69,
-	0x76, 0x61, 0x74, 0x65, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x7e, 0x0a, 0x0d, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53,
-	0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 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, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65,
-	0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x35, 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, 0x61,
-	0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x90, 0x01, 0x0a, 0x13, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72,
-	0x69, 0x7a, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x3a, 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, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69,
-	0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x3b, 0x2e, 0x63, 0x6f, 0x6e, 0x74,
+	0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6f,
+	0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12,
+	0x48, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x14, 0x20, 0x01, 0x28, 0x0e, 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, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55,
+	0x53, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x29, 0x0a, 0x10, 0x63, 0x6f, 0x6e,
+	0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x5f, 0x6c, 0x65, 0x66, 0x74, 0x18, 0x1e, 0x20,
+	0x01, 0x28, 0x05, 0x52, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73,
+	0x4c, 0x65, 0x66, 0x74, 0x22, 0x87, 0x02, 0x0a, 0x18, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f,
+	0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63,
+	0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x2c, 0x0a, 0x12, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74,
+	0x5f, 0x74, 0x63, 0x70, 0x5f, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x14, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x10, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x54, 0x63, 0x70, 0x41, 0x64, 0x64,
+	0x72, 0x65, 0x73, 0x73, 0x12, 0x26, 0x0a, 0x0f, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x74,
+	0x63, 0x70, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0d, 0x63,
+	0x6c, 0x69, 0x65, 0x6e, 0x74, 0x54, 0x63, 0x70, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x30, 0x0a, 0x14,
+	0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, 0x74, 0x63, 0x70, 0x5f, 0x61, 0x64, 0x64,
+	0x72, 0x65, 0x73, 0x73, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x12, 0x65, 0x6e, 0x64, 0x70,
+	0x6f, 0x69, 0x6e, 0x74, 0x54, 0x63, 0x70, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x2a,
+	0x0a, 0x11, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, 0x74, 0x63, 0x70, 0x5f, 0x70,
+	0x6f, 0x72, 0x74, 0x18, 0x32, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0f, 0x65, 0x6e, 0x64, 0x70, 0x6f,
+	0x69, 0x6e, 0x74, 0x54, 0x63, 0x70, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79,
+	0x70, 0x65, 0x18, 0x3c, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x22, 0x65,
+	0x0a, 0x19, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x48, 0x0a, 0x06, 0x73,
+	0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0e, 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, 0x43, 0x4f, 0x4e,
+	0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x52, 0x06, 0x73,
+	0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x93, 0x01, 0x0a, 0x1a, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43,
+	0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x44, 0x61, 0x74, 0x61, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69,
+	0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e,
+	0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x19, 0x0a, 0x08, 0x62, 0x79, 0x74,
+	0x65, 0x73, 0x5f, 0x75, 0x70, 0x18, 0x14, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x62, 0x79, 0x74,
+	0x65, 0x73, 0x55, 0x70, 0x12, 0x1d, 0x0a, 0x0a, 0x62, 0x79, 0x74, 0x65, 0x73, 0x5f, 0x64, 0x6f,
+	0x77, 0x6e, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x62, 0x79, 0x74, 0x65, 0x73, 0x44,
+	0x6f, 0x77, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e, 0x18, 0x28, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x06, 0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e, 0x22, 0x82, 0x01, 0x0a, 0x16,
+	0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x68, 0x0a, 0x12, 0x63, 0x6c, 0x6f, 0x73, 0x65, 0x5f,
+	0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x5f, 0x64, 0x61, 0x74, 0x61, 0x18, 0x0a, 0x20, 0x03,
+	0x28, 0x0b, 0x32, 0x3a, 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, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x44, 0x61, 0x74, 0x61, 0x52, 0x10,
+	0x63, 0x6c, 0x6f, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x44, 0x61, 0x74, 0x61,
+	0x22, 0x8c, 0x01, 0x0a, 0x1b, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63,
+	0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x44, 0x61, 0x74, 0x61,
+	0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69,
+	0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x48, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18,
+	0x14, 0x20, 0x01, 0x28, 0x0e, 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, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f,
+	0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22,
+	0x86, 0x01, 0x0a, 0x17, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x6b, 0x0a, 0x13, 0x63,
+	0x6c, 0x6f, 0x73, 0x65, 0x5f, 0x72, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x5f, 0x64, 0x61,
+	0x74, 0x61, 0x18, 0x0a, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x3b, 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, 0x6c, 0x6f, 0x73, 0x65, 0x43,
+	0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x44, 0x61, 0x74, 0x61, 0x52, 0x11, 0x63, 0x6c, 0x6f, 0x73, 0x65, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x44, 0x61, 0x74, 0x61, 0x32, 0xbe, 0x06, 0x0a, 0x0e, 0x53, 0x65, 0x73,
+	0x73, 0x69, 0x6f, 0x6e, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x7e, 0x0a, 0x0d, 0x4c,
+	0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 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, 0x4c, 0x6f,
+	0x6f, 0x6b, 0x75, 0x70, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65,
+	0x73, 0x74, 0x1a, 0x35, 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, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f,
+	0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x84, 0x01, 0x0a, 0x0f,
+	0x41, 0x63, 0x74, 0x69, 0x76, 0x61, 0x74, 0x65, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 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, 0x41, 0x63, 0x74, 0x69, 0x76, 0x61, 0x74, 0x65, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e,
+	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, 0x41, 0x63, 0x74, 0x69, 0x76, 0x61, 0x74,
+	0x65, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x00, 0x12, 0x7e, 0x0a, 0x0d, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73,
+	0x69, 0x6f, 0x6e, 0x12, 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, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73, 0x69,
+	0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x35, 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,
+	0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x43, 0x61, 0x6e, 0x63, 0x65,
-	0x72, 0x69, 0x7a, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65,
+	0x6c, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x8a, 0x01, 0x0a, 0x11, 0x43, 0x6f, 0x6e,
+	0x22, 0x00, 0x12, 0x90, 0x01, 0x0a, 0x13, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x65,
-	0x6e, 0x65, 0x63, 0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x38,
+	0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x3a, 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, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x3b, 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, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x8a, 0x01, 0x0a, 0x11, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63,
+	0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x38, 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, 0x6f, 0x6e,
+	0x6e, 0x65, 0x63, 0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x39, 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, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x43, 0x6f,
+	0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x00, 0x12, 0x84, 0x01, 0x0a, 0x0f, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e,
+	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 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, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e,
+	0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 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,
-	0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52,
-	0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x39, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72,
+	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x51, 0x5a, 0x4f, 0x67, 0x69, 0x74,
-	0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65,
+	0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72,
-	0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63,
+	0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72,
-	0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x6e, 0x61, 0x6c, 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c,
-	0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x84, 0x01, 0x0a, 0x0f, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43,
+	0x65, 0x72, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x69,
-	0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x36, 0x2e, 0x63, 0x6f, 0x6e, 0x74,
+	0x63, 0x65, 0x73, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72,
-	0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73,
+	0x6f, 0x74, 0x6f, 0x33,
-	0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x43, 0x6c, 0x6f, 0x73, 0x65,
-	0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 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, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69,
-	0x6f, 0x6e, 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/local/controller/servers/services/v1/session_service.proto

@@ -77,6 +77,7 @@ message CancelSessionResponse {
 
 message AuthorizeConnectionRequest {
 	string session_id = 10;
+	string worker_id = 20;
 }
 
 message AuthorizeConnectionResponse {

internal/servers/controller/handlers/workers/worker_service.go

@@ -52,15 +52,22 @@ var (
 )
 
 func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusRequest) (*pbs.StatusResponse, error) {
+	// TODO: on the worker, if we get errors back from this repeatedly, do we
+	// terminate all sessions since we can't know if they were canceled?
 	ws.logger.Trace("got status request from worker", "name", req.Worker.PrivateId, "address", req.Worker.Address, "jobs", req.GetJobs())
 	ws.updateTimes.Store(req.Worker.PrivateId, time.Now())
-	repo, err := ws.serversRepoFn()
+	serverRepo, err := ws.serversRepoFn()
 	if err != nil {
 		ws.logger.Error("error getting servers repo", "error", err)
 		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error aqcuiring repo to store worker status: %v", err)
 	}
+	sessRepo, err := ws.sessionRepoFn()
+	if err != nil {
+		ws.logger.Error("error getting sessions repo", "error", err)
+		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error aqcuiring repo to query session status: %v", err)
+	}
 	req.Worker.Type = resource.Worker.String()
-	controllers, _, err := repo.UpsertServer(ctx, req.Worker, servers.WithUpdateTags(req.GetUpdateTags()))
+	controllers, _, err := serverRepo.UpsertServer(ctx, req.Worker, servers.WithUpdateTags(req.GetUpdateTags()))
 	if err != nil {
 		ws.logger.Error("error storing worker status", "error", err)
 		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error storing worker status: %v", err)
@@ -69,15 +76,12 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 		Controllers: controllers,
 	}
 
-	// Happy path
+	// TODO (jeff, 05-2021): We should possibly list all connections here with
-	if len(req.GetJobs()) == 0 {
+	// their statuses, and check those against the found connections below. If
-		return ret, nil
+	// any we think should be closed are marked open by the worker, the worker
-	}
+	// should be told to close them.
 
-	sessRepo, err := ws.sessionRepoFn()
+	var foundConns []string
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error getting session repo: %v", err)
-	}
 
 	for _, jobStatus := range req.GetJobs() {
 		switch jobStatus.Job.GetType() {
@@ -87,12 +91,29 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 			if si == nil {
 				return nil, status.Error(codes.Internal, "Error getting session info at status time")
 			}
+
+			// Check connections before potentially bypassing the rest of the
+			// logic in the switch on si.Status.
+			sessConns := si.GetConnections()
+			for _, conn := range sessConns {
+				switch conn.Status {
+				case pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_AUTHORIZED,
+					pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED:
+					// If it's active, report it as found. Otherwise don't
+					// report as found, so that we should attempt to close it.
+					// Note that unspecified is the default state for the enum
+					// but it's not ever explicitly set by us.
+					foundConns = append(foundConns, conn.GetConnectionId())
+				}
+			}
+
 			switch si.Status {
 			case pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING,
 				pbs.SESSIONSTATUS_SESSIONSTATUS_TERMINATED:
 				// No need to see about canceling anything
 				continue
 			}
+
 			sessionId := si.GetSessionId()
 			sessionInfo, _, err := sessRepo.LookupSession(ctx, sessionId)
 			if err != nil {
@@ -131,6 +152,16 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 			}
 		}
 	}
+
+	// Run our connection cleanup function
+	closedConns, err := sessRepo.CloseDeadConnectionsOnWorkerReport(ctx, req.Worker.PrivateId, foundConns)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Error closing dead conns for worker %s: %v", req.Worker.PrivateId, err)
+	}
+	if closedConns > 0 {
+		ws.logger.Info("marked unclaimed connections as closed", "server_id", req.Worker.PrivateId, "count", closedConns)
+	}
+
 	return ret, nil
 }
 
@@ -302,7 +333,7 @@ func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs
 		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
 	}
 
-	connectionInfo, connStates, authzSummary, err := sessRepo.AuthorizeConnection(ctx, req.GetSessionId())
+	connectionInfo, connStates, authzSummary, err := sessRepo.AuthorizeConnection(ctx, req.GetSessionId(), req.GetWorkerId())
 	if err != nil {
 		return nil, err
 	}

internal/servers/worker/session.go

@@ -190,6 +190,7 @@ func (w *Worker) authorizeConnection(ctx context.Context, sessionId string) (*co
 
 	resp, err := conn.AuthorizeConnection(ctx, &pbs.AuthorizeConnectionRequest{
 		SessionId: sessionId,
+		WorkerId:  w.conf.RawConfig.Worker.Name,
 	})
 	if err != nil {
 		return nil, 0, fmt.Errorf("error authorizing connection: %w", err)

internal/session/connection.go

@@ -69,12 +69,12 @@ func NewConnection(sessionID, clientTcpAddress string, clientTcpPort uint32, end
 	return &c, nil
 }
 
-// AllocConnection will allocate a connection.
+// AllocConnection will allocate a Connection.
 func AllocConnection() Connection {
 	return Connection{}
 }
 
-// Clone creates a clone of the connection.
+// Clone creates a clone of the Connection.
 func (c *Connection) Clone() interface{} {
 	clone := &Connection{
 		PublicId:           c.PublicId,

internal/session/options.go

@@ -27,6 +27,8 @@ type options struct {
 	withTestTofu          []byte
 	withListingConvert    bool
 	withSessionIds        []string
+	withServerId          string
+	withDbOpts            []db.Option
 }
 
 func getDefaultOptions() options {
@@ -91,3 +93,18 @@ func withListingConvert(withListingConvert bool) Option {
 		o.withListingConvert = withListingConvert
 	}
 }
+
+// WithServerId allows the specification of the server id to use for the
+// operation.
+func WithServerId(id string) Option {
+	return func(o *options) {
+		o.withServerId = id
+	}
+}
+
+// WithDbOpts passes through given DB options to the DB layer
+func WithDbOpts(opts ...db.Option) Option {
+	return func(o *options) {
+		o.withDbOpts = opts
+	}
+}

internal/session/options_test.go

@@ -2,13 +2,11 @@ package session
 
 import (
 	"testing"
-	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 // Test_GetOpts provides unit tests for GetOpts and all the options
@@ -54,9 +52,8 @@ func Test_GetOpts(t *testing.T) {
 		assert.Equal(opts, testOpts)
 	})
 	t.Run("WithExpirationTime", func(t *testing.T) {
-		assert, require := assert.New(t), require.New(t)
+		assert := assert.New(t)
-		now, err := ptypes.TimestampProto(time.Now())
+		now := timestamppb.Now()
-		require.NoError(err)
 		opts := getOpts(WithExpirationTime(&timestamp.Timestamp{Timestamp: now}))
 		testOpts := getDefaultOptions()
 		testOpts.withExpirationTime = &timestamp.Timestamp{Timestamp: now}
@@ -78,4 +75,11 @@ func Test_GetOpts(t *testing.T) {
 		testOpts.withSessionIds = []string{"s_1", "s_2", "s_3"}
 		assert.Equal(opts, testOpts)
 	})
+	t.Run("WithServerId", func(t *testing.T) {
+		assert := assert.New(t)
+		opts := getOpts(WithServerId("worker1"))
+		testOpts := getDefaultOptions()
+		testOpts.withServerId = "worker1"
+		assert.Equal(opts, testOpts)
+	})
 }

internal/session/query.go

@@ -89,12 +89,14 @@ select * from terminated;
 	authorizeConnectionCte = `
 insert into session_connection (
 	session_id, 
-	public_id
+	public_id,
+	server_id
 )
 with active_session as ( 
 	select 
 		$1 as session_id,
-		$2 as public_id
+		$2 as public_id,
+		$3 as server_id
 	from
 		session s
 	where
@@ -232,4 +234,49 @@ where
     )
 )
 `
+
+	// connectionsToCloseCte finds connections that are:
+	//
+	// * not closed
+	// * not announced by a given server in its latest update
+	//
+	// and marks them as closed.
+	connectionsToCloseCte = `
+with
+  -- Find connections that are not closed so we can reference those IDs
+  unclosed_connections as (
+    select connection_id
+      from session_connection_state
+    where
+      -- It's the current state
+      end_time is null
+        and
+      -- Current state isn't closed state
+      state in ('authorized', 'connected')
+        and
+      -- It's not in limbo between when it moved into this state and when
+      -- it started being reported by the worker, which is roughly every
+      -- 2-3 seconds
+      start_time < now() - interval '10 seconds'
+  ),
+  connections_to_close as (
+    select public_id
+      from session_connection
+    where
+      -- Related to the worker that just reported to us
+      server_id = $1
+        and
+      -- These are connection IDs that just got reported to us by the given
+      -- worker, so they should not be considered closed.
+        %s
+      -- Only unclosed ones
+      public_id in (select connection_id from unclosed_connections)
+  )
+  update session_connection
+    set
+      closed_reason = 'system error'
+    where
+      public_id in (select public_id from connections_to_close)
+    returning public_id
+  `
 )

internal/session/repository_connection.go

@@ -3,13 +3,14 @@ package session
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/errors"
 )
 
 // LookupConnection will look up a connection in the repository and return the connection
-// with its states.  If the connection is not found, it will return nil, nil, nil.
+// with its states. If the connection is not found, it will return nil, nil, nil.
 // No options are currently supported.
 func (r *Repository) LookupConnection(ctx context.Context, connectionId string, _ ...Option) (*Connection, []*ConnectionState, error) {
 	const op = "session.(Repository).LookupConnection"
@@ -43,12 +44,21 @@ func (r *Repository) LookupConnection(ctx context.Context, connectionId string,
 	return &connection, states, nil
 }
 
-// ListConnections will sessions.  Supports the WithLimit and WithOrder options.
+// ListConnections will connections by session ID. If session ID is *, all
-func (r *Repository) ListConnections(ctx context.Context, sessionId string, opt ...Option) ([]*Connection, error) {
+// connections are returned. Supports the WithLimit and WithOrder options.
+func (r *Repository) ListConnectionsBySessionId(ctx context.Context, sessionId string, opt ...Option) ([]*Connection, error) {
 	const op = "session.(Repository).ListConnections"
 	opts := getOpts(opt...)
 	var connections []*Connection
-	err := r.list(ctx, &connections, "session_id = ?", []interface{}{sessionId}, opts) // pass options, so WithLimit and WithOrder are supported
+	var args []interface{}
+	var where string
+	if sessionId != "*" {
+		where = "session_id = ?"
+		args = append(args, sessionId)
+	}
+	// The where clause will be ignored if args is empty, if we don't want to
+	// scope to a specific session ID
+	err := r.list(ctx, &connections, where, args, opts) // pass options, so WithLimit and WithOrder are supported
 	if err != nil {
 		return nil, errors.Wrap(err, op)
 	}
@@ -95,6 +105,55 @@ func (r *Repository) DeleteConnection(ctx context.Context, publicId string, _ ..
 	return rowsDeleted, nil
 }
 
+// CloseDeadConnectionsOnWorkerReport will run the connectionsToClose CTE to
+// look for connections that should be marked closed because they are no longer
+// claimed by a server. This does notdetect connections where the server is no
+// longer reporting status; that's for a different CTE to be called by a
+// heartbeat detector.
+//
+// The foundConns input should be the currently-claimed connections; the CTE
+// uses a NOT IN clause to ensure these are excluded. It is not an error for
+// this to be empty as the worker could claim no connections; in that case all
+// connections will immediately transition to closed.
+func (r *Repository) CloseDeadConnectionsOnWorkerReport(ctx context.Context, serverId string, foundConns []string) (int, error) {
+	const op = "session.(Repository).CloseDeadConnectionsOnWorkerReport"
+	if serverId == "" {
+		return db.NoRowsAffected, errors.New(errors.InvalidParameter, op, "missing server id")
+	}
+
+	args := make([]interface{}, 0, len(foundConns)+1)
+	args = append(args, serverId)
+
+	var publicIdStr string
+	if len(foundConns) > 0 {
+		publicIdStr = `public_id not in (%s) and`
+		params := make([]string, len(foundConns))
+		for i, connId := range foundConns {
+			params[i] = fmt.Sprintf("$%d", i+2) // Add one for server ID, and offsets start at 1
+			args = append(args, connId)
+		}
+		publicIdStr = fmt.Sprintf(publicIdStr, strings.Join(params, ","))
+	}
+	var rowsAffected int
+	_, err := r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			var err error
+			rowsAffected, err = w.Exec(ctx, fmt.Sprintf(connectionsToCloseCte, publicIdStr), args)
+			if err != nil {
+				return errors.Wrap(err, op)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return db.NoRowsAffected, errors.Wrap(err, op)
+	}
+	return rowsAffected, nil
+}
+
 func fetchConnectionStates(ctx context.Context, r db.Reader, connectionId string, opt ...db.Option) ([]*ConnectionState, error) {
 	const op = "session.fetchConnectionStates"
 	var states []*ConnectionState

internal/session/repository_connection_test.go

@@ -5,12 +5,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/iam"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/oplog"
+	"github.com/hashicorp/boundary/sdk/strutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -93,7 +93,7 @@ func TestRepository_ListConnection(t *testing.T) {
 				testConnections = append(testConnections, c)
 			}
 			assert.Equal(tt.createCnt, len(testConnections))
-			got, err := repo.ListConnections(context.Background(), tt.args.searchForSessionId, tt.args.opt...)
+			got, err := repo.ListConnectionsBySessionId(context.Background(), tt.args.searchForSessionId, tt.args.opt...)
 			if tt.wantErr {
 				require.Error(err)
 				return
@@ -115,15 +115,13 @@ func TestRepository_ListConnection(t *testing.T) {
 				2222,
 			)
 		}
-		got, err := repo.ListConnections(context.Background(), session.PublicId, WithOrderByCreateTime(db.AscendingOrderBy))
+		got, err := repo.ListConnectionsBySessionId(context.Background(), session.PublicId, WithOrderByCreateTime(db.AscendingOrderBy))
 		require.NoError(err)
 		assert.Equal(wantCnt, len(got))
 
 		for i := 0; i < len(got)-1; i++ {
-			first, err := ptypes.Timestamp(got[i].CreateTime.Timestamp)
+			first := got[i].CreateTime.Timestamp.AsTime()
-			require.NoError(err)
+			second := got[i+1].CreateTime.Timestamp.AsTime()
-			second, err := ptypes.Timestamp(got[i+1].CreateTime.Timestamp)
-			require.NoError(err)
 			assert.True(first.Before(second))
 		}
 	})
@@ -211,3 +209,124 @@ func TestRepository_DeleteConnection(t *testing.T) {
 		})
 	}
 }
+
+func TestRepository_CloseDeadConnectionsOnWorkerReport(t *testing.T) {
+	t.Parallel()
+	require, assert := require.New(t), assert.New(t)
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	kms := kms.TestKms(t, conn, wrapper)
+	repo, err := NewRepository(rw, rw, kms)
+	require.NoError(err)
+	ctx := context.Background()
+	numConns := 12
+
+	// Create two "workers". One will remain untouched while the other "goes
+	// away and comes back" (worker 2).
+	worker1 := TestWorker(t, conn, wrapper, WithServerId("worker1"))
+	worker2 := TestWorker(t, conn, wrapper, WithServerId("worker2"))
+
+	// Create a few sessions on each, activate, and authorize a connection
+	var connIds []string
+	var worker2ConnIds []string
+	for i := 0; i < numConns; i++ {
+		serverId := worker1.PrivateId
+		if i%2 == 0 {
+			serverId = worker2.PrivateId
+		}
+		sess := TestDefaultSession(t, conn, wrapper, iamRepo, WithServerId(serverId), WithDbOpts(db.WithSkipVetForWrite(true)))
+		sess, _, err = repo.ActivateSession(ctx, sess.GetPublicId(), sess.Version, serverId, "worker", []byte("foo"))
+		require.NoError(err)
+		c, cs, _, err := repo.AuthorizeConnection(ctx, sess.GetPublicId(), serverId)
+		require.NoError(err)
+		require.Len(cs, 1)
+		require.Equal(StatusAuthorized, cs[0].Status)
+		connIds = append(connIds, c.GetPublicId())
+		if i%2 == 0 {
+			worker2ConnIds = append(worker2ConnIds, c.GetPublicId())
+		}
+	}
+
+	// Mark half of the connections connected and leave the others authorized.
+	// This is just to ensure we have a spread when we test it out.
+	for i, connId := range connIds {
+		if i%2 == 0 {
+			_, cs, err := repo.ConnectConnection(ctx, ConnectWith{
+				ConnectionId:       connId,
+				ClientTcpAddress:   "127.0.0.1",
+				ClientTcpPort:      22,
+				EndpointTcpAddress: "127.0.0.1",
+				EndpointTcpPort:    22,
+			})
+			require.NoError(err)
+			require.Len(cs, 2)
+			var foundAuthorized, foundConnected bool
+			for _, status := range cs {
+				if status.Status == StatusAuthorized {
+					foundAuthorized = true
+				}
+				if status.Status == StatusConnected {
+					foundConnected = true
+				}
+			}
+			require.True(foundAuthorized)
+			require.True(foundConnected)
+		}
+	}
+
+	// There is a 10 second delay to account for time for the connections to
+	// transition
+	time.Sleep(15 * time.Second)
+
+	// Now, advertise only some of the connection IDs for worker 2. After,
+	// all connection IDs for worker 1 should be showing as non-closed, and
+	// the ones for worker 2 not advertised should be closed.
+	shouldStayOpen := worker2ConnIds[0:2]
+	count, err := repo.CloseDeadConnectionsOnWorkerReport(ctx, worker2.GetPrivateId(), shouldStayOpen)
+	require.NoError(err)
+	assert.Equal(4, count)
+
+	// For the ones we didn't specify, we expect those to now be closed. We
+	// expect all others to be open.
+
+	shouldBeClosed := worker2ConnIds[2:]
+	conns, err := repo.ListConnectionsBySessionId(ctx, "*")
+	require.NoError(err)
+	for _, conn := range conns {
+		_, states, err := repo.LookupConnection(ctx, conn.PublicId)
+		require.NoError(err)
+		var foundClosed bool
+		for _, state := range states {
+			if state.Status == StatusClosed {
+				foundClosed = true
+				break
+			}
+		}
+		assert.True(foundClosed == strutil.StrListContains(shouldBeClosed, conn.PublicId))
+	}
+
+	// Now, advertise none of the connection IDs for worker 2. This is mainly to
+	// test that handling the case where we do not include IDs works properly as
+	// it changes the where clause.
+	count, err = repo.CloseDeadConnectionsOnWorkerReport(ctx, worker1.GetPrivateId(), nil)
+	require.NoError(err)
+	assert.Equal(6, count)
+
+	// We now expect all but those blessed few to be closed
+	conns, err = repo.ListConnectionsBySessionId(ctx, "*")
+	require.NoError(err)
+	for _, conn := range conns {
+		_, states, err := repo.LookupConnection(ctx, conn.PublicId)
+		require.NoError(err)
+		var foundClosed bool
+		for _, state := range states {
+			if state.Status == StatusClosed {
+				foundClosed = true
+				break
+			}
+		}
+		assert.True(foundClosed != strutil.StrListContains(shouldStayOpen, conn.PublicId))
+	}
+}

internal/session/repository_session.go

@@ -165,7 +165,7 @@ func (r *Repository) LookupSession(ctx context.Context, sessionId string, _ ...O
 	return &session, authzSummary, nil
 }
 
-// ListSessions will sessions.  Supports the WithLimit, WithScopeId and WithSessionIds options.
+// ListSessions will sessions.  Supports the WithLimit, WithScopeId, WithSessionIds, and WithServerId options.
 func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Session, error) {
 	const op = "session.(Repository).ListSessions"
 	opts := getOpts(opt...)
@@ -199,6 +199,10 @@ func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Sessio
 		}
 		where = append(where, fmt.Sprintf("s.public_id in (%s)", strings.Join(idsInClause, ",")))
 	}
+	if opts.withServerId != "" {
+		inClauseCnt += 1
+		where, args = append(where, fmt.Sprintf("server_id = $%d", inClauseCnt)), append(args, opts.withServerId)
+	}
 
 	var limit string
 	switch {
@@ -388,9 +392,9 @@ func (r *Repository) TerminateCompletedSessions(ctx context.Context) (int, error
 // * the hasn't expired based on the session.Expiration
 // * number of connections already created is less than session.ConnectionLimit
 // If authorization is success, it creates/stores a new connection in the repo
-// and returns it, along with it's states.  If the authorization fails, it
+// and returns it, along with its states.  If the authorization fails, it
 // an error with Code InvalidSessionState.
-func (r *Repository) AuthorizeConnection(ctx context.Context, sessionId string) (*Connection, []*ConnectionState, *ConnectionAuthzSummary, error) {
+func (r *Repository) AuthorizeConnection(ctx context.Context, sessionId, workerId string) (*Connection, []*ConnectionState, *ConnectionAuthzSummary, error) {
 	const op = "session.(Repository).AuthorizeConnection"
 	if sessionId == "" {
 		return nil, nil, nil, errors.Wrap(status.Error(codes.FailedPrecondition, "missing session id"), op, errors.WithCode(errors.InvalidParameter))
@@ -408,7 +412,7 @@ func (r *Repository) AuthorizeConnection(ctx context.Context, sessionId string)
 		db.StdRetryCnt,
 		db.ExpBackoff{},
 		func(reader db.Reader, w db.Writer) error {
-			rowsAffected, err := w.Exec(ctx, authorizeConnectionCte, []interface{}{sessionId, connectionId})
+			rowsAffected, err := w.Exec(ctx, authorizeConnectionCte, []interface{}{sessionId, connectionId, workerId})
 			if err != nil {
 				return errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("unable to authorize connection %s", sessionId)))
 			}

internal/session/repository_session_test.go

@@ -2,10 +2,10 @@ package session
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/hashicorp/boundary/internal/authtoken"
 	authtokenStore "github.com/hashicorp/boundary/internal/authtoken/store"
 	"github.com/hashicorp/boundary/internal/db"
@@ -15,6 +15,7 @@ import (
 	"github.com/hashicorp/boundary/internal/target"
 	targetStore "github.com/hashicorp/boundary/internal/target/store"
 	"github.com/lib/pq"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/iam"
@@ -129,10 +130,8 @@ func TestRepository_ListSession(t *testing.T) {
 		assert.Equal(wantCnt, len(got))
 
 		for i := 0; i < len(got)-1; i++ {
-			first, err := ptypes.Timestamp(got[i].CreateTime.Timestamp)
+			first := got[i].CreateTime.Timestamp.AsTime()
-			require.NoError(err)
+			second := got[i+1].CreateTime.Timestamp.AsTime()
-			second, err := ptypes.Timestamp(got[i+1].CreateTime.Timestamp)
-			require.NoError(err)
 			assert.True(first.Before(second))
 		}
 	})
@@ -189,6 +188,32 @@ func TestRepository_ListSession(t *testing.T) {
 		assert.Equal(StatusActive, got[0].States[0].Status)
 		assert.Equal(StatusPending, got[0].States[1].Status)
 	})
+	t.Run("withServerId", func(t *testing.T) {
+		assert, require := assert.New(t), require.New(t)
+		require.NoError(conn.Where("1=1").Delete(AllocSession()).Error)
+		for i := 0; i < 6; i++ {
+			if i%2 == 0 {
+				TestWorker(t, conn, wrapper, WithServerId(fmt.Sprintf("server-%d", i/2)))
+			}
+			_ = TestSession(t, conn, wrapper, composedOf,
+				WithServerId(fmt.Sprintf("server-%d", i/2)),
+				WithDbOpts(db.WithSkipVetForWrite(true)),
+			)
+		}
+		got, err := repo.ListSessions(context.Background())
+		require.NoError(err)
+		assert.Equal(6, len(got))
+
+		for i := 0; i < 3; i++ {
+			serverId := fmt.Sprintf("server-%d", i)
+			got, err = repo.ListSessions(context.Background(), WithServerId(serverId))
+			require.NoError(err)
+			assert.Equal(2, len(got))
+			for _, item := range got {
+				assert.Equal(serverId, item.ServerId)
+			}
+		}
+	})
 }
 
 func TestRepository_ListSessions_Multiple_Scopes(t *testing.T) {
@@ -488,6 +513,7 @@ func TestRepository_AuthorizeConnect(t *testing.T) {
 	repo, err := NewRepository(rw, rw, kms)
 	require.NoError(t, err)
 
+	var testServer string
 	setupFn := func(exp *timestamp.Timestamp) *Session {
 		composedOf := TestSessionParams(t, conn, wrapper, iamRepo)
 		if exp != nil {
@@ -495,6 +521,7 @@ func TestRepository_AuthorizeConnect(t *testing.T) {
 		}
 		s := TestSession(t, conn, wrapper, composedOf)
 		srv := TestWorker(t, conn, wrapper)
+		testServer = srv.PrivateId
 		tofu := TestTofu(t)
 		_, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, srv.Type, tofu)
 		require.NoError(t, err)
@@ -537,7 +564,7 @@ func TestRepository_AuthorizeConnect(t *testing.T) {
 		},
 		{
 			name:    "expired-session",
-			session: setupFn(&timestamp.Timestamp{Timestamp: ptypes.TimestampNow()}),
+			session: setupFn(&timestamp.Timestamp{Timestamp: timestamppb.Now()}),
 			wantErr: true,
 		},
 	}
@@ -545,7 +572,7 @@ func TestRepository_AuthorizeConnect(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
 
-			c, cs, authzInfo, err := repo.AuthorizeConnection(context.Background(), tt.session.PublicId)
+			c, cs, authzInfo, err := repo.AuthorizeConnection(context.Background(), tt.session.PublicId, testServer)
 			if tt.wantErr {
 				require.Error(err)
 				// TODO (jimlambrt 9/2020): add in tests for errorsIs once we
@@ -770,8 +797,7 @@ func TestRepository_TerminateCompletedSessions(t *testing.T) {
 
 	setupFn := func(limit int32, expireIn time.Duration, leaveOpen bool) *Session {
 		require.NotEqualf(t, int32(0), limit, "setupFn: limit cannot be zero")
-		exp, err := ptypes.TimestampProto(time.Now().Add(expireIn))
+		exp := timestamppb.New(time.Now().Add(expireIn))
-		require.NoError(t, err)
 		composedOf := TestSessionParams(t, conn, wrapper, iamRepo)
 		composedOf.ConnectionLimit = limit
 		composedOf.ExpirationTime = &timestamp.Timestamp{Timestamp: exp}
@@ -968,7 +994,7 @@ func TestRepository_TerminateCompletedSessions(t *testing.T) {
 					}
 					assert.Equal(args.wantTermed[found.PublicId].String(), found.TerminationReason)
 					t.Logf("terminated %s has a connection limit of %d", found.PublicId, found.ConnectionLimit)
-					conn, err := repo.ListConnections(context.Background(), found.PublicId)
+					conn, err := repo.ListConnectionsBySessionId(context.Background(), found.PublicId)
 					require.NoError(err)
 					for _, sc := range conn {
 						c, cs, err := repo.LookupConnection(context.Background(), sc.PublicId)
@@ -981,7 +1007,7 @@ func TestRepository_TerminateCompletedSessions(t *testing.T) {
 				} else {
 					t.Logf("not terminated %s has a connection limit of %d", found.PublicId, found.ConnectionLimit)
 					assert.Equal("", found.TerminationReason)
-					conn, err := repo.ListConnections(context.Background(), found.PublicId)
+					conn, err := repo.ListConnectionsBySessionId(context.Background(), found.PublicId)
 					require.NoError(err)
 					for _, sc := range conn {
 						cs, err := fetchConnectionStates(context.Background(), rw, sc.PublicId)

internal/session/session_test.go

@@ -5,13 +5,13 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/iam"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 func TestSession_Create(t *testing.T) {
@@ -21,9 +21,7 @@ func TestSession_Create(t *testing.T) {
 	iamRepo := iam.TestRepo(t, conn, wrapper)
 
 	composedOf := TestSessionParams(t, conn, wrapper, iamRepo)
-	future, err := ptypes.TimestampProto(time.Now().Add(time.Hour))
+	exp := &timestamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))}
-	require.NoError(t, err)
-	exp := &timestamp.Timestamp{Timestamp: future}
 
 	type args struct {
 		composedOf ComposedOf

internal/session/testing.go

@@ -6,7 +6,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/hashicorp/boundary/internal/auth/password"
 	"github.com/hashicorp/boundary/internal/authtoken"
 	"github.com/hashicorp/boundary/internal/db"
@@ -68,13 +67,14 @@ func TestState(t *testing.T, conn *gorm.DB, sessionId string, state Status) *Sta
 	return s
 }
 
-// TestSession creates a test session composed of c in the repository.
+// TestSession creates a test session composed of c in the repository. Options
+// are passed into New, and withServerId is handled locally.
 func TestSession(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper, c ComposedOf, opt ...Option) *Session {
 	t.Helper()
+	opts := getOpts(opt...)
 	require := require.New(t)
 	if c.ExpirationTime == nil {
-		future, err := ptypes.TimestampProto(time.Now().Add(time.Hour))
+		future := timestamppb.New(time.Now().Add(time.Hour))
-		require.NoError(err)
 		c.ExpirationTime = &timestamp.Timestamp{Timestamp: future}
 	}
 	rw := db.New(conn)
@@ -86,14 +86,15 @@ func TestSession(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper, c Compos
 	_, certBytes, err := newCert(wrapper, c.UserId, id, c.ExpirationTime.Timestamp.AsTime())
 	require.NoError(err)
 	s.Certificate = certBytes
+	s.ServerId = opts.withServerId
 
 	if len(s.TofuToken) != 0 {
 		err = s.encrypt(context.Background(), wrapper)
 	}
 	require.NoError(err)
-	err = rw.Create(context.Background(), s)
+	err = rw.Create(context.Background(), s, opts.withDbOpts...)
 	require.NoError(err)
-	ss, err := fetchStates(context.Background(), rw, s.PublicId, db.WithOrder("start_time desc"))
+	ss, err := fetchStates(context.Background(), rw, s.PublicId, append(opts.withDbOpts, db.WithOrder("start_time desc"))...)
 	require.NoError(err)
 	s.States = ss
 
@@ -103,12 +104,10 @@ func TestSession(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper, c Compos
 // TestDefaultSession creates a test session in the repository using defaults.
 func TestDefaultSession(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper, iamRepo *iam.Repository, opt ...Option) *Session {
 	t.Helper()
-	require := require.New(t)
 	composedOf := TestSessionParams(t, conn, wrapper, iamRepo)
-	future, err := ptypes.TimestampProto(time.Now().Add(time.Hour))
+	future := timestamppb.New(time.Now().Add(time.Hour))
-	require.NoError(err)
 	exp := &timestamp.Timestamp{Timestamp: future}
-	return TestSession(t, conn, wrapper, composedOf, WithExpirationTime(exp))
+	return TestSession(t, conn, wrapper, composedOf, append(opt, WithExpirationTime(exp))...)
 }
 
 // TestSessionParams returns an initialized ComposedOf which can be used to
@@ -169,17 +168,24 @@ func TestTofu(t *testing.T) []byte {
 	return []byte(tofu)
 }
 
-func TestWorker(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper) *servers.Server {
+// TestWorker inserts a worker into the db to satisfy foreign key constraints.
+// Supports the WithServerId option.
+func TestWorker(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper, opt ...Option) *servers.Server {
 	t.Helper()
 	rw := db.New(conn)
 	kms := kms.TestKms(t, conn, wrapper)
 	serversRepo, err := servers.NewRepository(rw, rw, kms)
 	require.NoError(t, err)
 
-	id, err := uuid.GenerateUUID()
+	opts := getOpts(opt...)
-	require.NoError(t, err)
+	id := opts.withServerId
+	if id == "" {
+		id, err = uuid.GenerateUUID()
+		require.NoError(t, err)
+		id = "test-session-worker-" + id
+	}
 	worker := &servers.Server{
-		PrivateId:   "test-session-worker-" + id,
+		PrivateId:   id,
 		Type:        servers.ServerTypeWorker.String(),
 		Description: "Test Session Worker",
 		Address:     "127.0.0.1",