diff --git a/internal/servers/controller/handlers/workers/worker_service.go b/internal/servers/controller/handlers/workers/worker_service.go index 6307b27ae..bb489a013 100644 --- a/internal/servers/controller/handlers/workers/worker_service.go +++ b/internal/servers/controller/handlers/workers/worker_service.go @@ -79,47 +79,16 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques Controllers: controllers, } - var ( - // For tracking the reported open connections. - reportedOpenConns []string - // For tracking the session IDs we've already requested - // cancellation for. We won't need to add connection cancel - // requests for these because canceling the session terminates the - // connections. - requestedSessionCancelIds []string - ) - - // This is a map of all sessions and their statuses. We keep track of - // this for easy lookup if we need to make change requests. - sessionStatuses := make(map[string]pbs.SESSIONSTATUS) + stateReport := make([]session.StateReport, 0, len(req.GetJobs())) for _, jobStatus := range req.GetJobs() { switch jobStatus.Job.GetType() { - // Check for session cancellation case pbs.JOBTYPE_JOBTYPE_SESSION: si := jobStatus.GetJob().GetSessionInfo() if si == nil { return nil, status.Error(codes.Internal, "Error getting session info at status time") } - // Record status. - sessionStatuses[si.GetSessionId()] = si.Status - - // 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. - reportedOpenConns = append(reportedOpenConns, conn.GetConnectionId()) - } - } - switch si.Status { case pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING, pbs.SESSIONSTATUS_SESSIONSTATUS_TERMINATED: @@ -127,77 +96,40 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques continue } - sessionId := si.GetSessionId() - sessionInfo, _, err := sessRepo.LookupSession(ctx, sessionId) - if err != nil { - return nil, status.Errorf(codes.Internal, "Error looking up session with id %s: %v", sessionId, err) + sr := session.StateReport{ + SessionId: si.GetSessionId(), + ConnectionIds: make([]string, 0, len(si.GetConnections())), } - if sessionInfo == nil { - return nil, status.Errorf(codes.Internal, "Unknown session ID %s at status time.", sessionId) - } - if len(sessionInfo.States) == 0 { - return nil, status.Error(codes.Internal, "Empty session states during lookup at status time.") - } - // If the session from the DB is in canceling status, and we're - // here, it means the job is in pending or active; cancel it. If - // it's in terminated status something went wrong and we're - // mismatched, so ensure we cancel it also. - currState := sessionInfo.States[0].Status - if currState.ProtoVal() != si.Status { - switch currState { - case session.StatusCanceling, - session.StatusTerminated: - // If we're here the job is pending or active so we do want - // to actually send a change request - ret.JobsRequests = append(ret.JobsRequests, &pbs.JobChangeRequest{ - Job: &pbs.Job{ - Type: pbs.JOBTYPE_JOBTYPE_SESSION, - JobInfo: &pbs.Job_SessionInfo{ - SessionInfo: &pbs.SessionJobInfo{ - SessionId: sessionId, - Status: currState.ProtoVal(), - }, - }, - }, - RequestType: pbs.CHANGETYPE_CHANGETYPE_UPDATE_STATE, - }) - // Log the session ID so we don't add a duplicate change - // request on connection normalization. - requestedSessionCancelIds = append(requestedSessionCancelIds, sessionId) + for _, conn := range si.GetConnections() { + switch conn.Status { + case pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_AUTHORIZED, + pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED: + sr.ConnectionIds = append(sr.ConnectionIds, conn.GetConnectionId()) } } + stateReport = append(stateReport, sr) } } - // Normalize the current state of connections on the worker side - // with the data from the controller. In other words, if one of our - // found connections isn't supposed to be alive still, kill it. - // - // This is separate from the above session normalization and is - // additive to it, we don't add sessions that have already been - // added there as canceling sessions already closes the - // connections. - shouldCloseConnections, err := connectionRepo.ShouldCloseConnectionsOnWorker(ctx, reportedOpenConns, requestedSessionCancelIds) + notActive, err := session.WorkerStatusReport(ctx, sessRepo, connectionRepo, req.Worker.PrivateId, stateReport) if err != nil { - return nil, status.Errorf(codes.Internal, "Error fetching connections that should be closed: %v", err) + return nil, status.Errorf(codes.Internal, "Error comparing state of sessions for worker: %s: %v", req.Worker.PrivateId, err) } - - for sessionId, connIds := range shouldCloseConnections { + for _, na := range notActive { var connChanges []*pbs.Connection - for _, connId := range connIds { + for _, connId := range na.ConnectionIds { connChanges = append(connChanges, &pbs.Connection{ ConnectionId: connId, Status: session.StatusClosed.ProtoVal(), }) } - ret.JobsRequests = append(ret.JobsRequests, &pbs.JobChangeRequest{ Job: &pbs.Job{ Type: pbs.JOBTYPE_JOBTYPE_SESSION, JobInfo: &pbs.Job_SessionInfo{ SessionInfo: &pbs.SessionJobInfo{ - SessionId: sessionId, - Status: sessionStatuses[sessionId], + SessionId: na.SessionId, + Status: na.Status.ProtoVal(), Connections: connChanges, }, }, @@ -206,15 +138,6 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques }) } - // Run our controller-side cleanup function. - closedConns, err := connectionRepo.CloseDeadConnectionsForWorker(ctx, req.Worker.PrivateId, reportedOpenConns) - if err != nil { - return nil, status.Errorf(codes.Internal, "Error closing dead conns for worker %s: %v", req.Worker.PrivateId, err) - } - if closedConns > 0 { - event.WriteSysEvent(ctx, op, "marked unclaimed connections as closed", "server_id", req.Worker.PrivateId, "count", closedConns) - } - return ret, nil } diff --git a/internal/session/query.go b/internal/session/query.go index 901b01f6d..39066ac80 100644 --- a/internal/session/query.go +++ b/internal/session/query.go @@ -294,51 +294,6 @@ where ); ` - // closeDeadConnectionsCte finds connections that are: - // - // * not closed - // * not announced by a given server in its latest update - // - // and marks them as closed. - closeDeadConnectionsCte = ` -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 < wt_sub_seconds_from_now(@worker_state_delay_seconds) - ), - connections_to_close as ( - select public_id - from session_connection - where - -- Related to the worker that just reported to us - server_id = @server_id - 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; -` - // closeConnectionsForDeadServersCte finds connections that are: // // * not closed @@ -374,34 +329,48 @@ with order by closed_connections.server_id; ` - // shouldCloseConnectionsCte finds connections that are marked as closed in - // the database given a set of connection IDs. They are returned along with - // their associated session ID. - // - // The second parameter is a set of session IDs that we have already - // submitted a session-wide close request for, so sending another change - // request for them would be redundant. - shouldCloseConnectionsCte = ` + orphanedConnectionsCte = ` +-- Find connections that are not closed so we can reference those IDs with - -- Find connections that are closed so we can reference those IDs - closed_connections as ( + unclosed_connections as ( select connection_id from session_connection_state where -- It's the current state end_time is null - and - -- Current state is closed - state = 'closed' - and - connection_id in (%s) + -- Current state isn't closed state + and state in ('authorized', 'connected') + -- 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 + and start_time < wt_sub_seconds_from_now(@worker_state_delay_seconds) + ), + connections_to_close as ( + select public_id + from session_connection + where + -- Related to the worker that just reported to us + server_id = @server_id + -- Only unclosed ones + and public_id in (select connection_id from unclosed_connections) + -- These are connection IDs that just got reported to us by the given + -- worker, so they should not be considered closed. + %s ) - select public_id, session_id - from session_connection - where - public_id in (select connection_id from closed_connections) - -- Below fmt arg is filled in if there are session IDs to filter against - %s +update session_connection + set + closed_reason = 'system error' + where + public_id in (select public_id from connections_to_close) +returning public_id; +` + checkIfNotActive = ` +select session_id, state + from session_state ss +where + (ss.state = 'canceling' or ss.state = 'terminated') + and ss.end_time is null + %s ; ` ) diff --git a/internal/session/repository_connection.go b/internal/session/repository_connection.go index e63b0bd20..4f384f900 100644 --- a/internal/session/repository_connection.go +++ b/internal/session/repository_connection.go @@ -350,54 +350,6 @@ func (r *ConnectionRepository) DeleteConnection(ctx context.Context, publicId st return rowsDeleted, nil } -// CloseDeadConnectionsForWorker will run closeDeadConnectionsCte to look for -// connections that should be marked closed because they are no longer claimed -// by a server. -// -// 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 *ConnectionRepository) CloseDeadConnectionsForWorker(ctx context.Context, serverId string, foundConns []string) (int, error) { - const op = "session.(ConnectionRepository).CloseDeadConnectionsForWorker" - if serverId == "" { - return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "missing server id") - } - - args := make([]interface{}, 0, len(foundConns)+2) - args = append(args, sql.Named("server_id", serverId)) - args = append(args, sql.Named("worker_state_delay_seconds", r.workerStateDelay.Seconds())) - - 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) - args = append(args, sql.Named(fmt.Sprintf("%d", i), 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(closeDeadConnectionsCte, publicIdStr), args) - if err != nil { - return errors.Wrap(ctx, err, op) - } - return nil - }, - ) - if err != nil { - return db.NoRowsAffected, errors.Wrap(ctx, err, op) - } - return rowsAffected, nil -} - type CloseConnectionsForDeadWorkersResult struct { ServerId string LastUpdateTime time.Time @@ -448,67 +400,52 @@ func (r *ConnectionRepository) CloseConnectionsForDeadWorkers(ctx context.Contex return results, nil } -// ShouldCloseConnectionsOnWorker will run shouldCloseConnectionsCte to look -// for connections that the worker should close because they are currently -// reporting them as open incorrectly. -// -// The foundConns input here is used to filter closed connection states. This -// is further filtered against the filterSessions input, which is expected to -// be a set of sessions we've already submitted close requests for, so adding -// them again would be redundant. -// -// The returned map[string][]string is indexed by session ID. -func (r *ConnectionRepository) ShouldCloseConnectionsOnWorker(ctx context.Context, foundConns, filterSessions []string) (map[string][]string, error) { - const op = "session.(ConnectionRepository).ShouldCloseConnectionsOnWorker" - if len(foundConns) < 1 { - return nil, nil // nothing to do - } +// closeOrphanedConnections looks for connections that are still active, but where not reported by the worker. +func (r *ConnectionRepository) closeOrphanedConnections(ctx context.Context, workerId string, reportedConnections []string) ([]string, error) { + const op = "session.(ConnectionRepository).closeOrphanedConnections" - args := make([]interface{}, 0, len(foundConns)+len(filterSessions)) + var orphanedConns []string - // foundConns first - connsParams := make([]string, len(foundConns)) - for i, connId := range foundConns { - connsParams[i] = fmt.Sprintf("@%d", i+1) - args = append(args, sql.Named(fmt.Sprintf("%d", i+1), connId)) - } - connsStr := strings.Join(connsParams, ",") + args := make([]interface{}, 0, len(reportedConnections)+2) + args = append(args, sql.Named("server_id", workerId)) + args = append(args, sql.Named("worker_state_delay_seconds", r.workerStateDelay.Seconds())) - // then filterSessions - var sessionsStr string - if len(filterSessions) > 0 { - offset := len(foundConns) + 1 - sessionsParams := make([]string, len(filterSessions)) - for i, sessionId := range filterSessions { - sessionsParams[i] = fmt.Sprintf("@%d", i+offset) - args = append(args, sql.Named(fmt.Sprintf("%d", i+offset), sessionId)) + var notInClause string + if len(reportedConnections) > 0 { + notInClause = `and public_id not in (%s)` + params := make([]string, len(reportedConnections)) + for i, connId := range reportedConnections { + params[i] = fmt.Sprintf("@%d", i) + args = append(args, sql.Named(fmt.Sprintf("%d", i), connId)) } - - const sessionIdFmtStr = `and session_id not in (%s)` - sessionsStr = fmt.Sprintf(sessionIdFmtStr, strings.Join(sessionsParams, ",")) + notInClause = fmt.Sprintf(notInClause, strings.Join(params, ",")) } - rows, err := r.reader.Query( + _, err := r.writer.DoTx( ctx, - fmt.Sprintf(shouldCloseConnectionsCte, connsStr, sessionsStr), - args, + db.StdRetryCnt, + db.ExpBackoff{}, + func(reader db.Reader, w db.Writer) error { + rows, err := r.reader.Query(ctx, fmt.Sprintf(orphanedConnectionsCte, notInClause), args) + if err != nil { + return errors.Wrap(ctx, err, op) + } + defer rows.Close() + + for rows.Next() { + var connectionId string + if err := rows.Scan(&connectionId); err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("scan row failed")) + } + orphanedConns = append(orphanedConns, connectionId) + } + return nil + }, ) if err != nil { - return nil, errors.Wrap(ctx, err, op) + return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error comparing state")) } - defer rows.Close() - - result := make(map[string][]string) - for rows.Next() { - var connectionId, sessionId string - if err := rows.Scan(&connectionId, &sessionId); err != nil { - return nil, errors.Wrap(ctx, err, op) - } - - result[sessionId] = append(result[sessionId], connectionId) - } - - return result, nil + return orphanedConns, nil } func fetchConnectionStates(ctx context.Context, r db.Reader, connectionId string, opt ...db.Option) ([]*ConnectionState, error) { diff --git a/internal/session/repository_connection_test.go b/internal/session/repository_connection_test.go index d31ea1524..10ffb29ec 100644 --- a/internal/session/repository_connection_test.go +++ b/internal/session/repository_connection_test.go @@ -12,7 +12,6 @@ import ( "github.com/hashicorp/boundary/internal/kms" "github.com/hashicorp/boundary/internal/oplog" "github.com/hashicorp/boundary/internal/servers" - "github.com/hashicorp/go-secure-stdlib/strutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -342,7 +341,7 @@ func TestRepository_DeleteConnection(t *testing.T) { } } -func TestRepository_CloseDeadConnectionsOnWorker(t *testing.T) { +func TestRepository_orphanedConnections(t *testing.T) { t.Parallel() ctx := context.Background() require, assert := require.New(t), assert.New(t) @@ -352,7 +351,7 @@ func TestRepository_CloseDeadConnectionsOnWorker(t *testing.T) { iamRepo := iam.TestRepo(t, conn, wrapper) kms := kms.TestKms(t, conn, wrapper) repo, err := NewRepository(rw, rw, kms) - connRepo, err := NewConnectionRepository(ctx, rw, rw, kms) + connRepo, err := NewConnectionRepository(ctx, rw, rw, kms, WithWorkerStateDelay(0)) require.NoError(err) numConns := 12 @@ -363,6 +362,7 @@ func TestRepository_CloseDeadConnectionsOnWorker(t *testing.T) { // Create a few sessions on each, activate, and authorize a connection var connIds []string + var worker1ConnIds []string var worker2ConnIds []string for i := 0; i < numConns; i++ { serverId := worker1.PrivateId @@ -379,6 +379,8 @@ func TestRepository_CloseDeadConnectionsOnWorker(t *testing.T) { connIds = append(connIds, c.GetPublicId()) if i%2 == 0 { worker2ConnIds = append(worker2ConnIds, c.GetPublicId()) + } else { + worker1ConnIds = append(worker1ConnIds, c.GetPublicId()) } } @@ -410,59 +412,28 @@ func TestRepository_CloseDeadConnectionsOnWorker(t *testing.T) { } } - // 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 := connRepo.CloseDeadConnectionsForWorker(ctx, worker2.GetPrivateId(), shouldStayOpen) + found, err := connRepo.closeOrphanedConnections(ctx, worker2.GetPrivateId(), shouldStayOpen) require.NoError(err) - assert.Equal(4, count) + fmt.Printf("shouldstate: %v\nfound: %v\n", shouldStayOpen, found) + require.Equal(4, len(found)) // For the ones we didn't specify, we expect those to now be closed. We // expect all others to be open. - shouldBeClosed := worker2ConnIds[2:] - var conns []*Connection - require.NoError(repo.list(ctx, &conns, "", nil)) - for _, conn := range conns { - _, states, err := connRepo.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)) - } + shouldBeFound := worker2ConnIds[2:] + assert.ElementsMatch(found, shouldBeFound) // 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 = connRepo.CloseDeadConnectionsForWorker(ctx, worker1.GetPrivateId(), nil) + found, err = connRepo.closeOrphanedConnections(ctx, worker1.GetPrivateId(), nil) require.NoError(err) - assert.Equal(6, count) - - // We now expect all but those blessed few to be closed - conns = nil - require.NoError(repo.list(ctx, &conns, "", nil)) - for _, conn := range conns { - _, states, err := connRepo.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)) - } + assert.Equal(6, len(found)) + assert.ElementsMatch(found, worker1ConnIds) } func TestRepository_CloseConnectionsForDeadWorkers(t *testing.T) { @@ -748,155 +719,6 @@ func TestRepository_CloseConnectionsForDeadWorkers(t *testing.T) { } } -func TestRepository_ShouldCloseConnectionsOnWorker(t *testing.T) { - t.Parallel() - ctx := context.Background() - require := require.New(t) - conn, _ := db.TestSetup(t, "postgres") - rw := db.New(conn) - wrapper := db.TestWrapper(t) - iamRepo := iam.TestRepo(t, conn, wrapper) - kms := kms.TestKms(t, conn, wrapper) - repo, err := NewRepository(rw, rw, kms) - connRepo, err := NewConnectionRepository(ctx, rw, rw, kms) - require.NoError(err) - numConns := 12 - - // Create a worker, we only need one here as our query is dependent - // on connection and not worker. - worker1 := TestWorker(t, conn, wrapper, WithServerId("worker1")) - - // Create a few sessions on each, activate, and authorize a connection - var connIds []string - sessionConnIds := make(map[string][]string) - for i := 0; i < numConns; i++ { - serverId := worker1.PrivateId - sess := TestDefaultSession(t, conn, wrapper, iamRepo, WithServerId(serverId), WithDbOpts(db.WithSkipVetForWrite(true))) - sessionId := sess.GetPublicId() - sess, _, err = repo.ActivateSession(ctx, sessionId, sess.Version, serverId, "worker", []byte("foo")) - require.NoError(err) - c, cs, err := connRepo.AuthorizeConnection(ctx, sess.GetPublicId(), serverId) - require.NoError(err) - require.Len(cs, 1) - require.Equal(StatusAuthorized, cs[0].Status) - connId := c.GetPublicId() - connIds = append(connIds, connId) - sessionConnIds[sessionId] = append(sessionConnIds[sessionId], connId) - } - - // Mark half of the connections connected, close the other half. - for i, connId := range connIds { - if i%2 == 0 { - _, cs, err := connRepo.ConnectConnection(ctx, ConnectWith{ - ConnectionId: connId, - ClientTcpAddress: "127.0.0.1", - ClientTcpPort: 22, - EndpointTcpAddress: "127.0.0.1", - EndpointTcpPort: 22, - UserClientIp: "127.0.0.1", - }) - 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) - } else { - resp, err := connRepo.closeConnections(ctx, []CloseWith{ - { - ConnectionId: connId, - ClosedReason: ConnectionCanceled, - }, - }) - require.NoError(err) - require.Len(resp, 1) - cs := resp[0].ConnectionStates - require.Len(cs, 2) - var foundAuthorized, foundClosed bool - for _, status := range cs { - if status.Status == StatusAuthorized { - foundAuthorized = true - } - if status.Status == StatusClosed { - foundClosed = true - } - } - require.True(foundAuthorized) - require.True(foundClosed) - } - } - - // There is a 10 second delay to account for time for the connections to - // transition - time.Sleep(15 * time.Second) - - // Now we try some scenarios. - { - // First test an empty set. - result, err := connRepo.ShouldCloseConnectionsOnWorker(ctx, nil, nil) - require.NoError(err) - require.Zero(result, "should be empty when no connections are supplied") - } - - { - // Here we pass in all of our connections without a filter on - // session. This should return half of the connections - the ones - // that we marked as closed. - // - // Create a copy of our session map with the sessions that have - // closed connections taken out. - expectedSessionConnIds := make(map[string][]string) - for sessionId, connIds := range sessionConnIds { - for _, connId := range connIds { - if testIsConnectionClosed(ctx, t, connRepo, connId) { - expectedSessionConnIds[sessionId] = append(expectedSessionConnIds[sessionId], connId) - } - } - } - - // Send query, use all connections w/o a filter on sessions. - actualSessionConnIds, err := connRepo.ShouldCloseConnectionsOnWorker(ctx, connIds, nil) - require.NoError(err) - require.Equal(expectedSessionConnIds, actualSessionConnIds) - } - - { - // Finally, add a session filter. We do this by just alternating - // the session IDs we want to filter on. - expectedSessionConnIds := make(map[string][]string) - var filterSessionIds []string - var filterSession bool - for sessionId, connIds := range sessionConnIds { - for _, connId := range connIds { - if testIsConnectionClosed(ctx, t, connRepo, connId) { - if !filterSession { - expectedSessionConnIds[sessionId] = append(expectedSessionConnIds[sessionId], connId) - } else { - filterSessionIds = append(filterSessionIds, sessionId) - } - - // Toggle filterSession here (instead of just outer session - // loop) so that we aren't just lining up on - // connected/disconnected connections. - filterSession = !filterSession - } - } - } - - // Send query with the session filter. - actualSessionConnIds, err := connRepo.ShouldCloseConnectionsOnWorker(ctx, connIds, filterSessionIds) - require.NoError(err) - require.Equal(expectedSessionConnIds, actualSessionConnIds) - } -} - func TestRepository_CloseConnections(t *testing.T) { t.Parallel() ctx := context.Background() @@ -978,12 +800,3 @@ func TestRepository_CloseConnections(t *testing.T) { }) } } - -func testIsConnectionClosed(ctx context.Context, t *testing.T, repo *ConnectionRepository, connId string) bool { - require := require.New(t) - _, states, err := repo.LookupConnection(ctx, connId) - require.NoError(err) - // Use first state as this LookupConnections returns ordered by - // start time, descending. - return states[0].Status == StatusClosed -} diff --git a/internal/session/repository_session.go b/internal/session/repository_session.go index 4ecc5c6f7..c1d2f37a2 100644 --- a/internal/session/repository_session.go +++ b/internal/session/repository_session.go @@ -588,6 +588,59 @@ func (r *Repository) updateState(ctx context.Context, sessionId string, sessionV return &updatedSession, returnedStates, nil } +// checkIfNoLongerActive checks the given sessions to see if they are in a +// non-active state, i.e. "canceling" or "terminated" +// It returns a []StateReport for each session that is not active, with its current status. +func (r *Repository) checkIfNoLongerActive(ctx context.Context, reportedSessions []string) ([]StateReport, error) { + const op = "session.(Repository).checkIfNotActive" + + notActive := make([]StateReport, 0, len(reportedSessions)) + args := make([]interface{}, 0, len(reportedSessions)) + var inClause string + + if len(reportedSessions) <= 0 { + return notActive, nil + } + + inClause = `and session_id in (%s)` + params := make([]string, len(reportedSessions)) + for i, sessId := range reportedSessions { + params[i] = fmt.Sprintf("@%d", i) + args = append(args, sql.Named(fmt.Sprintf("%d", i), sessId)) + } + inClause = fmt.Sprintf(inClause, strings.Join(params, ",")) + + _, err := r.writer.DoTx( + ctx, + db.StdRetryCnt, + db.ExpBackoff{}, + func(reader db.Reader, w db.Writer) error { + rows, err := r.reader.Query(ctx, fmt.Sprintf(checkIfNotActive, inClause), args) + if err != nil { + return errors.Wrap(ctx, err, op) + } + defer rows.Close() + + for rows.Next() { + var sessionId string + var status Status + if err := rows.Scan(&sessionId, &status); err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("scan row failed")) + } + notActive = append(notActive, StateReport{ + SessionId: sessionId, + Status: status, + }) + } + return nil + }, + ) + if err != nil { + return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error checking if sessions are no longer active")) + } + return notActive, nil +} + func fetchStates(ctx context.Context, r db.Reader, sessionId string, opt ...db.Option) ([]*State, error) { const op = "session.fetchStates" var states []*State diff --git a/internal/session/service_worker_status_report.go b/internal/session/service_worker_status_report.go new file mode 100644 index 000000000..0cc62a426 --- /dev/null +++ b/internal/session/service_worker_status_report.go @@ -0,0 +1,48 @@ +package session + +import ( + "context" + "fmt" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/observability/event" +) + +// StateReport is used to report on the state of a Session. +type StateReport struct { + SessionId string + Status Status + ConnectionIds []string +} + +// WorkerStatusReport is a domain service function that compares the state of +// sessions and connections as reported by a Worker, to the known state in the +// repositories. It returns a []StateReport for each session that is in the +// canceling or terminated state. It also will check for any orphaned +// connections, which is defined as a connection that is in an active state, +// but was not reported by worker. Any orphaned connections will be marked as +// closed. +func WorkerStatusReport(ctx context.Context, repo *Repository, connRepo *ConnectionRepository, workerId string, report []StateReport) ([]StateReport, error) { + const op = "session.WorkerStatusReport" + + reportedConnections := make([]string, 0) + reportedSessions := make([]string, 0, len(report)) + for _, r := range report { + reportedSessions = append(reportedSessions, r.SessionId) + reportedConnections = append(reportedConnections, r.ConnectionIds...) + } + + notActive, err := repo.checkIfNoLongerActive(ctx, reportedSessions) + if err != nil { + return nil, errors.New(ctx, errors.Internal, op, fmt.Sprintf("Error checking session state for worker %s: %v", workerId, err)) + } + + closed, err := connRepo.closeOrphanedConnections(ctx, workerId, reportedConnections) + if err != nil { + return notActive, errors.New(ctx, errors.Internal, op, fmt.Sprintf("Error closing orphaned connections for worker %s: %v", workerId, err)) + } + if len(closed) > 0 { + event.WriteSysEvent(ctx, op, "marked unclaimed connections as closed", "server_id", workerId, "count", len(closed)) + } + return notActive, err +} diff --git a/internal/session/service_worker_status_report_test.go b/internal/session/service_worker_status_report_test.go new file mode 100644 index 000000000..0589bdea3 --- /dev/null +++ b/internal/session/service_worker_status_report_test.go @@ -0,0 +1,394 @@ +package session_test + +import ( + "context" + "testing" + + "github.com/hashicorp/boundary/internal/authtoken" + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/host/static" + "github.com/hashicorp/boundary/internal/iam" + "github.com/hashicorp/boundary/internal/kms" + "github.com/hashicorp/boundary/internal/servers" + "github.com/hashicorp/boundary/internal/session" + "github.com/hashicorp/boundary/internal/target" + "github.com/hashicorp/boundary/internal/target/tcp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestWorkerStatusReport(t *testing.T) { + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) + + serverRepo, _ := servers.NewRepository(rw, rw, kms) + serverRepo.UpsertServer(ctx, &servers.Server{ + PrivateId: "test_controller1", + Type: "controller", + Address: "127.0.0.1", + }) + serverRepo.UpsertServer(ctx, &servers.Server{ + PrivateId: "test_worker1", + Type: "worker", + Address: "127.0.0.1", + }) + + repo, err := session.NewRepository(rw, rw, kms) + require.NoError(t, err) + connRepo, err := session.NewConnectionRepository(ctx, rw, rw, kms, session.WithWorkerStateDelay(0)) + require.NoError(t, err) + + at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId()) + uId := at.GetIamUserId() + hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0] + hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0] + h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0] + static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h}) + tar := tcp.TestTarget( + ctx, + t, conn, prj.GetPublicId(), "test", + target.WithHostSources([]string{hs.GetPublicId()}), + target.WithSessionConnectionLimit(-1), + ) + + type testCase struct { + worker *servers.Server + req []session.StateReport + want []session.StateReport + orphanedConnections []string + } + cases := []struct { + name string + caseFn func(t *testing.T) testCase + }{ + { + name: "No Sessions", + caseFn: func(t *testing.T) testCase { + worker := session.TestWorker(t, conn, wrapper) + return testCase{ + worker: worker, + req: []session.StateReport{}, + want: []session.StateReport{}, + } + }, + }, + { + name: "No Sessions already canceled", + caseFn: func(t *testing.T) testCase { + worker := session.TestWorker(t, conn, wrapper) + sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu := session.TestTofu(t) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + require.NoError(t, err) + require.NoError(t, err) + + _, _, err = connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) + require.NoError(t, err) + + _, err = repo.CancelSession(ctx, sess.PublicId, sess.Version) + require.NoError(t, err) + + return testCase{ + worker: worker, + req: []session.StateReport{}, + want: []session.StateReport{}, + } + }, + }, + { + name: "Still Active", + caseFn: func(t *testing.T) testCase { + worker := session.TestWorker(t, conn, wrapper) + sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu := session.TestTofu(t) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + require.NoError(t, err) + require.NoError(t, err) + + connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) + require.NoError(t, err) + return testCase{ + worker: worker, + req: []session.StateReport{ + { + SessionId: sess.PublicId, + Status: session.StatusActive, + ConnectionIds: []string{connection.PublicId}, + }, + }, + want: []session.StateReport{}, + } + }, + }, + { + name: "SessionClosed", + caseFn: func(t *testing.T) testCase { + worker := session.TestWorker(t, conn, wrapper) + sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu := session.TestTofu(t) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + require.NoError(t, err) + connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) + require.NoError(t, err) + _, err = repo.CancelSession(ctx, sess.PublicId, sess.Version) + require.NoError(t, err) + + return testCase{ + worker: worker, + req: []session.StateReport{ + { + SessionId: sess.PublicId, + Status: session.StatusActive, + ConnectionIds: []string{connection.PublicId}, + }, + }, + want: []session.StateReport{ + { + SessionId: sess.PublicId, + Status: session.StatusCanceling, + }, + }, + } + }, + }, + { + name: "MultipleSessionsClosed", + caseFn: func(t *testing.T) testCase { + worker := session.TestWorker(t, conn, wrapper) + sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu := session.TestTofu(t) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + require.NoError(t, err) + connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) + require.NoError(t, err) + _, err = repo.CancelSession(ctx, sess.PublicId, sess.Version) + require.NoError(t, err) + + sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu2 := session.TestTofu(t) + sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, worker.Type, tofu2) + require.NoError(t, err) + connection2, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker.PrivateId) + require.NoError(t, err) + _, err = repo.CancelSession(ctx, sess2.PublicId, sess2.Version) + require.NoError(t, err) + + return testCase{ + worker: worker, + req: []session.StateReport{ + { + SessionId: sess.PublicId, + Status: session.StatusActive, + ConnectionIds: []string{connection.PublicId}, + }, + { + SessionId: sess2.PublicId, + Status: session.StatusActive, + ConnectionIds: []string{connection2.PublicId}, + }, + }, + want: []session.StateReport{ + { + SessionId: sess.PublicId, + Status: session.StatusCanceling, + }, + { + SessionId: sess2.PublicId, + Status: session.StatusCanceling, + }, + }, + } + }, + }, + { + name: "OrphanedConnection", + caseFn: func(t *testing.T) testCase { + worker := session.TestWorker(t, conn, wrapper) + sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu := session.TestTofu(t) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + require.NoError(t, err) + connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) + require.NoError(t, err) + + sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu2 := session.TestTofu(t) + sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, worker.Type, tofu2) + require.NoError(t, err) + connection2, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker.PrivateId) + require.NoError(t, err) + require.NotEqual(t, connection.PublicId, connection2.PublicId) + + return testCase{ + worker: worker, + req: []session.StateReport{ + { + SessionId: sess2.PublicId, + Status: session.StatusActive, + ConnectionIds: []string{connection2.PublicId}, + }, + }, + want: []session.StateReport{}, + orphanedConnections: []string{connection.PublicId}, + } + }, + }, + { + name: "MultipleSessionsAndOrphanedConnections", + caseFn: func(t *testing.T) testCase { + worker := session.TestWorker(t, conn, wrapper) + sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu := session.TestTofu(t) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + require.NoError(t, err) + connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) + require.NoError(t, err) + _, err = repo.CancelSession(ctx, sess.PublicId, sess.Version) + require.NoError(t, err) + + sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ScopeId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + ConnectionLimit: 10, + }) + tofu2 := session.TestTofu(t) + sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, worker.Type, tofu2) + require.NoError(t, err) + connection2, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker.PrivateId) + require.NoError(t, err) + connection3, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker.PrivateId) + require.NoError(t, err) + _, err = repo.CancelSession(ctx, sess2.PublicId, sess2.Version) + require.NoError(t, err) + + return testCase{ + worker: worker, + req: []session.StateReport{ + { + SessionId: sess.PublicId, + Status: session.StatusActive, + ConnectionIds: []string{connection.PublicId}, + }, + { + SessionId: sess2.PublicId, + Status: session.StatusActive, + ConnectionIds: []string{connection2.PublicId}, + }, + }, + want: []session.StateReport{ + { + SessionId: sess.PublicId, + Status: session.StatusCanceling, + }, + { + SessionId: sess2.PublicId, + Status: session.StatusCanceling, + }, + }, + orphanedConnections: []string{connection3.PublicId}, + } + }, + }, + } + + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + + tc := tt.caseFn(t) + + got, err := session.WorkerStatusReport(ctx, repo, connRepo, tc.worker.PrivateId, tc.req) + require.NoError(err) + assert.ElementsMatch(tc.want, got) + for _, dc := range tc.orphanedConnections { + gotConn, states, err := connRepo.LookupConnection(ctx, dc) + require.NoError(err) + assert.Equal(session.ConnectionSystemError, session.ClosedReason(gotConn.ClosedReason)) + assert.Equal(2, len(states)) + assert.Nil(states[0].EndTime) + assert.Equal(session.StatusClosed, states[0].Status) + } + }) + } +} diff --git a/internal/tests/cluster/session_cleanup_test.go b/internal/tests/cluster/session_cleanup_test.go index a9dae2484..eeb3e5ca5 100644 --- a/internal/tests/cluster/session_cleanup_test.go +++ b/internal/tests/cluster/session_cleanup_test.go @@ -35,21 +35,14 @@ import ( // worker is managing the lifecycle of a connection and will properly // unclaim it closed once the connection resumes, ensuring the // connection is marked as closed on the worker. -// -// * controller: Here, the controller is the one doing the work. The -// connection will be open on the worker until status checks resume -// from the worker. At this point, the controller will request the -// status change on the worker, physically closing the connection -// there. type timeoutBurdenType string const ( - timeoutBurdenTypeDefault timeoutBurdenType = "default" - timeoutBurdenTypeWorker timeoutBurdenType = "worker" - timeoutBurdenTypeController timeoutBurdenType = "controller" + timeoutBurdenTypeDefault timeoutBurdenType = "default" + timeoutBurdenTypeWorker timeoutBurdenType = "worker" ) -var timeoutBurdenCases = []timeoutBurdenType{timeoutBurdenTypeDefault, timeoutBurdenTypeWorker, timeoutBurdenTypeController} +var timeoutBurdenCases = []timeoutBurdenType{timeoutBurdenTypeDefault, timeoutBurdenTypeWorker} func controllerGracePeriod(ty timeoutBurdenType) time.Duration { if ty == timeoutBurdenTypeWorker { @@ -60,10 +53,6 @@ func controllerGracePeriod(ty timeoutBurdenType) time.Duration { } func workerGracePeriod(ty timeoutBurdenType) time.Duration { - if ty == timeoutBurdenTypeController { - return helper.DefaultGracePeriod * 10 - } - return helper.DefaultGracePeriod } @@ -172,11 +161,6 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin sess.ExpectConnectionStateOnWorker(ctx, t, w1, session.StatusClosed) sess.ExpectConnectionStateOnController(ctx, t, c1.Controller().ConnectionRepoFn, session.StatusConnected) - case timeoutBurdenTypeController: - // Wait on controller, then check worker - sess.ExpectConnectionStateOnController(ctx, t, c1.Controller().ConnectionRepoFn, session.StatusClosed) - sess.ExpectConnectionStateOnWorker(ctx, t, w1, session.StatusConnected) - default: // Should be closed on both worker and controller. Wait on // worker then check controller. @@ -184,16 +168,7 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin sess.ExpectConnectionStateOnController(ctx, t, c1.Controller().ConnectionRepoFn, session.StatusClosed) } - // Run send/receive test again to check expected connection-level - // behavior - if burdenCase == timeoutBurdenTypeController { - // Burden on controller, should be successful until connection - // resumes - sConn.TestSendRecvAll(t) - } else { - // Connection should die in other cases - sConn.TestSendRecvFail(t) - } + sConn.TestSendRecvFail(t) // Resume the connection, and reconnect. event.WriteSysEvent(ctx, op, "resuming controller/worker link") @@ -212,12 +187,6 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin // connections are actually closed now that the worker is // properly reporting in again. sess.ExpectConnectionStateOnController(ctx, t, c1.Controller().ConnectionRepoFn, session.StatusClosed) - - case timeoutBurdenTypeController: - // If we are expecting the controller to be the source of - // truth, the connection should now be forcibly closed after - // the worker gets a status change request back. - sConn.TestSendRecvFail(t) } // Proceed with new connection test @@ -373,11 +342,6 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing sess.ExpectConnectionStateOnWorker(ctx, t, w1, session.StatusClosed) sess.ExpectConnectionStateOnController(ctx, t, c1.Controller().ConnectionRepoFn, session.StatusConnected) - case timeoutBurdenTypeController: - // Wait on controller, then check worker - sess.ExpectConnectionStateOnController(ctx, t, c1.Controller().ConnectionRepoFn, session.StatusClosed) - sess.ExpectConnectionStateOnWorker(ctx, t, w1, session.StatusConnected) - default: // Should be closed on both worker and controller. Wait on // worker then check controller. @@ -387,14 +351,7 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing // Run send/receive test again to check expected connection-level // behavior - if burdenCase == timeoutBurdenTypeController { - // Burden on controller, should be successful until connection - // resumes - sConn.TestSendRecvAll(t) - } else { - // Connection should die in other cases - sConn.TestSendRecvFail(t) - } + sConn.TestSendRecvFail(t) // Finally resume both, try again. Should behave as per normal. event.WriteSysEvent(ctx, op, "resuming connections to both controllers") @@ -414,12 +371,6 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing // connections are actually closed now that the worker is // properly reporting in again. sess.ExpectConnectionStateOnController(ctx, t, c1.Controller().ConnectionRepoFn, session.StatusClosed) - - case timeoutBurdenTypeController: - // If we are expecting the controller to be the source of - // truth, the connection should now be forcibly closed after - // the worker gets a status change request back. - sConn.TestSendRecvFail(t) } // Proceed with new connection test