Perf and stability improve for lib/srv/regular

Parallelized and increased frequency of checks for many tests.

Stabilized the following tests.

* TestAllowedLabels
* TestDirectTCPIP
* TestLockInForce
* TestTCPIPForward
* TestAgentForward
* TestX11Forward

Original flaky behavior for the above tests can be reproduced using
[stress](https://pkg.go.dev/golang.org/x/tools/cmd/stress) with the
following command:

```
go test -o flaky.test ./... && \
  stress -p 30 -failfast ./flaky.test -test.run="TestName"
```

The following tests still show up when running the full test suite for
`lib/srv/regular` under `stress`.

* TestMultipleExecCommands
* TestSessionAuditLog

Author: russjones

lib/client/client.go

@@ -931,14 +931,26 @@ func (c *NodeClient) remoteListenAndForward(ctx context.Context, ln net.Listener
 		"remote_addr", remoteAddr,
 	)
 	log.InfoContext(ctx, "Starting remote port forwarding")
+	defer log.InfoContext(ctx, "Shutting down remote port forwarding", "error", ctx.Err())
 
-	for ctx.Err() == nil {
+	for {
 		conn, err := acceptWithContext(ctx, ln)
 		if err != nil {
-			if ctx.Err() == nil {
+			switch {
+			// Caller closed context to stop forwarding. For example the user
+			// ran "tsh ssh -N -R" then hit Ctrl-C.
+			case ctx.Err() != nil:
+				return
+			// The remote server closed the connection. For example, client_idle_timeout
+			// was hit.
+			case errors.Is(err, io.EOF):
+				return
+			// Accepting forwarded connection failed, but listener may still
+			// accept. For example, a transient network issue.
+			default:
 				log.ErrorContext(ctx, "Remote port forwarding failed", "error", err)
+				continue
 			}
-			continue
 		}
 
 		go func() {
@@ -947,7 +959,6 @@ func (c *NodeClient) remoteListenAndForward(ctx context.Context, ln net.Listener
 			}
 		}()
 	}
-	log.InfoContext(ctx, "Shutting down remote port forwarding", "error", ctx.Err())
 }
 
 // GetRemoteTerminalSize fetches the terminal size of a given SSH session.

lib/srv/regular/sshserver.go

@@ -2460,20 +2460,49 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co
 	if err != nil {
 		return trace.Wrap(err)
 	}
-	defer scx.Close()
 	listener, err := s.listenTCPIP(ctx, scx, scx.SrcAddr)
 	if err != nil {
+		if serr := scx.Close(); serr != nil {
+			s.logger.DebugContext(ctx, "Failed while cleaning up request",
+				"request_type", teleport.TCPIPForwardRequest,
+				"server_context_close_error", serr,
+				"error", err)
+		}
 		return trace.Wrap(err)
 	}
 
 	// If the client didn't request a specific port, the chosen port needs to
 	// be reported back.
-	srcHost, _, err := sshutils.SplitHostPort(scx.SrcAddr)
+	srcHost, srcPort, err := sshutils.SplitHostPort(scx.SrcAddr)
 	if err != nil {
+		if lerr := listener.Close(); lerr != nil {
+			s.logger.DebugContext(ctx, "Failed while cleaning up request",
+				"request_type", teleport.TCPIPForwardRequest,
+				"listener_close_error", lerr,
+				"error", err)
+		}
+		if serr := scx.Close(); serr != nil {
+			s.logger.DebugContext(ctx, "Failed while cleaning up request",
+				"request_type", teleport.TCPIPForwardRequest,
+				"server_context_close_error", serr,
+				"error", err)
+		}
 		return trace.Wrap(err)
 	}
 	_, listenPort, err := sshutils.SplitHostPort(listener.Addr().String())
 	if err != nil {
+		if lerr := listener.Close(); lerr != nil {
+			s.logger.DebugContext(ctx, "Failed while cleaning up request",
+				"request_type", teleport.TCPIPForwardRequest,
+				"listener_close_error", lerr,
+				"error", err)
+		}
+		if serr := scx.Close(); serr != nil {
+			s.logger.DebugContext(ctx, "Failed while cleaning up request",
+				"request_type", teleport.TCPIPForwardRequest,
+				"server_context_close_error", serr,
+				"error", err)
+		}
 		return trace.Wrap(err)
 	}
 	scx.SrcAddr = sshutils.JoinHostPort(srcHost, listenPort)
@@ -2483,6 +2512,7 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co
 
 	// spawn remote forwarding handler to multiplex connections to the forwarded port
 	go func() {
+		defer scx.Close()
 		stopEvent := scx.GetPortForwardEvent(events.PortForwardRemoteEvent, events.PortForwardStopCode, scx.SrcAddr)
 		defer s.emitAuditEventWithLog(ctx, &stopEvent)
 
@@ -2525,6 +2555,7 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co
 				logger.WarnContext(ctx, "failed to open channel", "error", err)
 				continue
 			}
+			ch = scx.TrackActivity(ch)
 			go ssh.DiscardRequests(rch)
 			go io.Copy(io.Discard, ch.Stderr())
 			go func() {
@@ -2548,11 +2579,7 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co
 	// Report addr back to the client.
 	if r.WantReply {
 		var payload []byte
-		req, err := sshutils.ParseTCPIPForwardReq(r.Payload)
-		if err != nil {
-			return trace.Wrap(err)
-		}
-		if req.Port == 0 {
+		if srcPort == 0 {
 			payload = ssh.Marshal(struct {
 				Port uint32
 			}{Port: uint32(listener.Addr().(*net.TCPAddr).Port)})