Merge pull request #1910 from hashicorp/hugoamvieira-worker-start-listeners

refactor(worker): Startup, shutdown and cluster tests changes

internal/cmd/commands/dev/dev.go

@@ -659,7 +659,7 @@ func (c *Command) Run(args []string) int {
 
 		if err := c.worker.Start(); err != nil {
 			retErr := fmt.Errorf("Error starting worker: %w", err)
-			if err := c.worker.Shutdown(false); err != nil {
+			if err := c.worker.Shutdown(); err != nil {
 				c.UI.Error(retErr.Error())
 				retErr = fmt.Errorf("Error shutting down worker: %w", err)
 			}
@@ -709,7 +709,7 @@ func (c *Command) Run(args []string) int {
 			}
 
 			if !c.flagControllerOnly {
-				if err := c.worker.Shutdown(false); err != nil {
+				if err := c.worker.Shutdown(); err != nil {
 					c.UI.Error(fmt.Errorf("Error shutting down worker: %w", err).Error())
 				}
 			}

internal/cmd/commands/server/server.go

@@ -636,7 +636,7 @@ func (c *Command) StartWorker() error {
 
 	if err := c.worker.Start(); err != nil {
 		retErr := fmt.Errorf("Error starting worker: %w", err)
-		if err := c.worker.Shutdown(false); err != nil {
+		if err := c.worker.Shutdown(); err != nil {
 			c.UI.Error(retErr.Error())
 			retErr = fmt.Errorf("Error shutting down worker: %w", err)
 		}
@@ -677,7 +677,7 @@ func (c *Command) WaitForInterrupt() int {
 
 			// Do worker shutdown
 			if c.Config.Worker != nil {
-				if err := c.worker.Shutdown(false); err != nil {
+				if err := c.worker.Shutdown(); err != nil {
 					c.UI.Error(fmt.Errorf("Error shutting down worker: %w", err).Error())
 				}
 			}

internal/servers/worker/controller_connection.go

@@ -138,17 +138,13 @@ func (w *Worker) createClientConn(addr string) error {
 func (w *Worker) workerAuthTLSConfig() (*tls.Config, *base.WorkerAuthInfo, error) {
 	var err error
 	info := &base.WorkerAuthInfo{
-		Name:            w.conf.RawConfig.Worker.Name,
-		Description:     w.conf.RawConfig.Worker.Description,
-		ConnectionNonce: w.testReusedAuthNonce,
+		Name:        w.conf.RawConfig.Worker.Name,
+		Description: w.conf.RawConfig.Worker.Description,
 	}
-	if info.ConnectionNonce == "" {
-		if info.ConnectionNonce, err = base62.Random(20); err != nil {
-			return nil, nil, err
-		}
-		if w.testReuseAuthNonces {
-			w.testReusedAuthNonce = info.ConnectionNonce
-		}
+
+	info.ConnectionNonce, err = w.nonceFn(20)
+	if err != nil {
+		return nil, nil, err
 	}
 
 	pubKey, privKey, err := ed25519.GenerateKey(w.conf.SecureRandomReader)

internal/servers/worker/listeners.go

@@ -5,12 +5,13 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"log"
 	"net"
 	"net/http"
 	"os"
-	"sync"
 	"time"
 
+	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/libs/alpnmux"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/go-multierror"
@@ -18,7 +19,7 @@ import (
 
 func (w *Worker) startListeners() error {
 	const op = "worker.(Worker).startListeners"
-	servers := make([]func(), 0, len(w.conf.Listeners))
+
 	e := event.SysEventer()
 	if e == nil {
 		return fmt.Errorf("%s: sys eventer not initialized", op)
@@ -27,70 +28,15 @@ func (w *Worker) startListeners() error {
 	if err != nil {
 		return fmt.Errorf("%s: unable to initialize std logger: %w", op, err)
 	}
-	for _, ln := range w.conf.Listeners {
-		for _, purpose := range ln.Config.Purpose {
-			switch purpose {
-			case "api", "cluster", "ops":
-				// We may have this in dev mode; ignore
-				continue
 
-			case "proxy":
-				// Do nothing; handle below
-
-			default:
-				return fmt.Errorf("unknown listener purpose %q", purpose)
-			}
-
-			handler, err := w.handler(HandlerProperties{
-				ListenerConfig: ln.Config,
-			})
-			if err != nil {
-				return fmt.Errorf("%s: %w", op, err)
-			}
-
-			cancelCtx := w.baseContext
-
-			server := &http.Server{
-				Handler:           handler,
-				ReadHeaderTimeout: 10 * time.Second,
-				ReadTimeout:       30 * time.Second,
-				ErrorLog:          logger,
-				BaseContext: func(net.Listener) context.Context {
-					return cancelCtx
-				},
-			}
-			ln.HTTPServer = server
-
-			if ln.Config.HTTPReadHeaderTimeout > 0 {
-				server.ReadHeaderTimeout = ln.Config.HTTPReadHeaderTimeout
-			}
-			if ln.Config.HTTPReadTimeout > 0 {
-				server.ReadTimeout = ln.Config.HTTPReadTimeout
-			}
-			if ln.Config.HTTPWriteTimeout > 0 {
-				server.WriteTimeout = ln.Config.HTTPWriteTimeout
-			}
-			if ln.Config.HTTPIdleTimeout > 0 {
-				server.IdleTimeout = ln.Config.HTTPIdleTimeout
-			}
-
-			// Clear out in case this is a second start of the controller
-			ln.Mux.UnregisterProto(alpnmux.DefaultProto)
-			ln.Mux.UnregisterProto(alpnmux.NoProto)
-			l, err := ln.Mux.RegisterProto(alpnmux.DefaultProto, &tls.Config{
-				GetConfigForClient: w.getSessionTls,
-			})
-			if err != nil {
-				return fmt.Errorf("error getting tls listener: %w", err)
-			}
-			if l == nil {
-				return errors.New("could not get tls listener")
-			}
-
-			servers = append(servers, func() {
-				go server.Serve(l)
-			})
+	servers := make([]func(), 0, len(w.listeners))
+	for i := range w.listeners {
+		ln := w.listeners[i]
+		workerServer, err := w.configureForWorker(ln, logger)
+		if err != nil {
+			return fmt.Errorf("%s: failed to configure for worker: %w", op, err)
 		}
+		servers = append(servers, workerServer)
 	}
 
 	for _, s := range servers {
@@ -100,38 +46,127 @@ func (w *Worker) startListeners() error {
 	return nil
 }
 
-func (w *Worker) stopListeners() error {
-	serverWg := new(sync.WaitGroup)
-	for _, ln := range w.conf.Listeners {
-		localLn := ln
-		serverWg.Add(1)
-		go func() {
-			defer serverWg.Done()
-
-			shutdownKill, shutdownKillCancel := context.WithTimeout(w.baseContext, localLn.Config.MaxRequestDuration)
-			defer shutdownKillCancel()
-
-			if localLn.HTTPServer != nil {
-				_ = localLn.HTTPServer.Shutdown(shutdownKill)
-			}
-		}()
+func (w *Worker) configureForWorker(ln *base.ServerListener, log *log.Logger) (func(), error) {
+	handler, err := w.handler(HandlerProperties{ListenerConfig: ln.Config})
+	if err != nil {
+		return nil, err
 	}
-	serverWg.Wait()
 
-	var retErr *multierror.Error
-	if !w.conf.RawConfig.DevController {
-		for _, ln := range w.conf.Listeners {
-			if err := ln.Mux.Close(); err != nil {
-				if _, ok := err.(*os.PathError); ok && ln.Config.Type == "unix" {
-					// The rmListener probably tried to remove the file but it
-					// didn't exist, ignore the error; this is a conflict
-					// between rmListener and the default Go behavior of
-					// removing auto-vivified Unix domain sockets.
-				} else {
-					retErr = multierror.Append(retErr, err)
-				}
-			}
+	cancelCtx := w.baseContext
+	server := &http.Server{
+		Handler:           handler,
+		ReadHeaderTimeout: 10 * time.Second,
+		ReadTimeout:       30 * time.Second,
+		ErrorLog:          log,
+		BaseContext: func(net.Listener) context.Context {
+			return cancelCtx
+		},
+	}
+	ln.HTTPServer = server
+
+	if ln.Config.HTTPReadHeaderTimeout > 0 {
+		server.ReadHeaderTimeout = ln.Config.HTTPReadHeaderTimeout
+	}
+	if ln.Config.HTTPReadTimeout > 0 {
+		server.ReadTimeout = ln.Config.HTTPReadTimeout
+	}
+	if ln.Config.HTTPWriteTimeout > 0 {
+		server.WriteTimeout = ln.Config.HTTPWriteTimeout
+	}
+	if ln.Config.HTTPIdleTimeout > 0 {
+		server.IdleTimeout = ln.Config.HTTPIdleTimeout
+	}
+
+	// Clear out in case this is a second start of the controller
+	ln.Mux.UnregisterProto(alpnmux.DefaultProto)
+	ln.Mux.UnregisterProto(alpnmux.NoProto)
+	l, err := ln.Mux.RegisterProto(alpnmux.DefaultProto, &tls.Config{
+		GetConfigForClient: w.getSessionTls,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error getting tls listener: %w", err)
+	}
+	if l == nil {
+		return nil, errors.New("could not get tls listener")
+	}
+
+	return func() { go server.Serve(l) }, nil
+}
+
+func (w *Worker) stopServersAndListeners() error {
+	var closeErrors *multierror.Error
+	err := w.stopHttpServersAndListeners()
+	if err != nil {
+		closeErrors = multierror.Append(closeErrors, err)
+	}
+
+	err = w.stopAnyListeners()
+	if err != nil {
+		closeErrors = multierror.Append(closeErrors, err)
+	}
+
+	return closeErrors.ErrorOrNil()
+}
+
+func (w *Worker) stopHttpServersAndListeners() error {
+	var closeErrors *multierror.Error
+	for i := range w.listeners {
+		ln := w.listeners[i]
+		if ln.HTTPServer == nil {
+			continue
+		}
+
+		ctx, cancel := context.WithTimeout(w.baseContext, ln.Config.MaxRequestDuration)
+		ln.HTTPServer.Shutdown(ctx)
+		cancel()
+
+		err := ln.Mux.Close()
+		err = listenerCloseErrorCheck(ln.Config.Type, err)
+		if err != nil {
+			multierror.Append(closeErrors, err)
 		}
 	}
-	return retErr.ErrorOrNil()
+
+	return closeErrors.ErrorOrNil()
+}
+
+// stopAnyListeners does a final once over the known
+// listeners to make sure we didn't miss any;
+// expected to run at the end of stopServersAndListeners.
+func (w *Worker) stopAnyListeners() error {
+	var closeErrors *multierror.Error
+	for _, ln := range w.listeners {
+		if ln == nil || ln.Mux == nil {
+			continue
+		}
+
+		err := ln.Mux.Close()
+		err = listenerCloseErrorCheck(ln.Config.Type, err)
+		if err != nil {
+			multierror.Append(closeErrors, err)
+		}
+	}
+
+	return closeErrors.ErrorOrNil()
+}
+
+// listenerCloseErrorCheck does some validation on an error returned
+// by a net.Listener's Close function, and ignores a few cases
+// where we don't actually want an error to be returned.
+func listenerCloseErrorCheck(lnType string, err error) error {
+	if errors.Is(err, net.ErrClosed) {
+		// Ignore net.ErrClosed - The listener was already closed,
+		// so there's nothing else to do.
+		return nil
+	}
+	if _, ok := err.(*os.PathError); ok && lnType == "unix" {
+		// The underlying rmListener probably tried to remove
+		// the file but it didn't exist, ignore the error;
+		// this is a conflict between rmListener and the
+		// default Go behavior of removing auto-vivified
+		// Unix domain sockets.
+		return nil
+	}
+
+	return err
 }

internal/servers/worker/listeners_test.go

@@ -0,0 +1,389 @@
+package worker
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/boundary/internal/cmd/base"
+	"github.com/hashicorp/boundary/internal/libs/alpnmux"
+	"github.com/hashicorp/go-secure-stdlib/listenerutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStartListeners(t *testing.T) {
+	tests := []struct {
+		name       string
+		listeners  []*listenerutil.ListenerConfig
+		expErr     bool
+		expErrMsg  string
+		assertions func(t *testing.T, w *Worker, addrs []string)
+	}{
+		{
+			name: "one tcp listener, one unix listener",
+			listeners: []*listenerutil.ListenerConfig{
+				{
+					Type:       "tcp",
+					Purpose:    []string{"proxy"},
+					Address:    "127.0.0.1:0",
+					TLSDisable: true,
+				},
+				{
+					Type:       "unix",
+					Purpose:    []string{"proxy"},
+					Address:    "/tmp/boundary-worker-test-start-listeners.sock",
+					TLSDisable: true,
+				},
+			},
+			expErr: false,
+			assertions: func(t *testing.T, w *Worker, addrs []string) {
+				require.Len(t, addrs, 2)
+
+				_, err := http.Get("http://" + addrs[0] + "/v1/proxy/")
+				require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
+
+				cl := http.Client{
+					Transport: &http.Transport{
+						Dial: func(network, addr string) (net.Conn, error) { return net.Dial("unix", addrs[1]) },
+					},
+				}
+				_, err = cl.Get("http://anything.domain/v1/proxy/")
+				require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
+			},
+		},
+		{
+			name: "multiple tcp listeners, multiple unix listeners",
+			listeners: []*listenerutil.ListenerConfig{
+				{
+					Type:       "tcp",
+					Purpose:    []string{"proxy"},
+					Address:    "127.0.0.1:0",
+					TLSDisable: true,
+				},
+				{
+					Type:       "tcp",
+					Purpose:    []string{"proxy"},
+					Address:    "127.0.0.1:0",
+					TLSDisable: true,
+				},
+				{
+					Type:       "unix",
+					Purpose:    []string{"proxy"},
+					Address:    "/tmp/boundary-worker-test-start-listeners0.sock",
+					TLSDisable: true,
+				},
+				{
+					Type:       "unix",
+					Purpose:    []string{"proxy"},
+					Address:    "/tmp/boundary-worker-test-start-listeners1.sock",
+					TLSDisable: true,
+				},
+			},
+			expErr: false,
+			assertions: func(t *testing.T, w *Worker, addrs []string) {
+				require.Len(t, addrs, 4)
+
+				_, err := http.Get("http://" + addrs[0] + "/v1/proxy/")
+				require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
+
+				_, err = http.Get("http://" + addrs[1] + "/v1/proxy/")
+				require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
+
+				for _, proxyAddr := range []string{addrs[2], addrs[3]} {
+					cl := http.Client{
+						Transport: &http.Transport{
+							Dial: func(network, addr string) (net.Conn, error) { return net.Dial("unix", proxyAddr) },
+						},
+					}
+					_, err = cl.Get("http://anything.domain/v1/proxy")
+					require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation
+				}
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			conf := NewTestWorker(t, &TestWorkerOpts{DisableAutoStart: true}).w.conf
+			conf.RawConfig.SharedConfig.Listeners = tt.listeners
+
+			err := conf.SetupListeners(nil, conf.RawConfig.SharedConfig, []string{"proxy"})
+			require.NoError(t, err)
+
+			w, err := New(conf)
+			require.NoError(t, err)
+			w.baseContext = context.Background()
+
+			err = w.startListeners()
+			if tt.expErr {
+				require.EqualError(t, err, tt.expErrMsg)
+				return
+			}
+
+			require.NoError(t, err)
+
+			addrs := make([]string, 0)
+			for _, l := range w.listeners {
+				addrs = append(addrs, l.Mux.Addr().String())
+			}
+			if tt.assertions != nil {
+				tt.assertions(t, w, addrs)
+			}
+		})
+	}
+}
+
+func TestStopHttpServersAndListeners(t *testing.T) {
+	tests := []struct {
+		name       string
+		workerFn   func(t *testing.T) *Worker
+		assertions func(t *testing.T, w *Worker)
+		expErr     bool
+		expErrStr  string
+	}{
+		{
+			name: "no listeners",
+			workerFn: func(t *testing.T) *Worker {
+				return &Worker{listeners: []*base.ServerListener{}}
+			},
+			expErr: false,
+		},
+		{
+			name: "nil listeners slice",
+			workerFn: func(t *testing.T) *Worker {
+				return &Worker{listeners: nil}
+			},
+			expErr: false,
+		},
+		{
+			name: "listeners with nil http server",
+			workerFn: func(t *testing.T) *Worker {
+				return &Worker{
+					listeners: []*base.ServerListener{
+						{HTTPServer: nil},
+						{HTTPServer: nil},
+						{HTTPServer: nil},
+					},
+				}
+			},
+			expErr: false,
+		},
+		{
+			name: "multiple listeners",
+			workerFn: func(t *testing.T) *Worker {
+				l1, err := net.Listen("tcp", "127.0.0.1:0")
+				require.NoError(t, err)
+				l2, err := net.Listen("unix", "/tmp/boundary-worker-TestStopHttpServersAndListeners-"+strconv.FormatInt(time.Now().UnixNano(), 10))
+				require.NoError(t, err)
+
+				s1 := &http.Server{}
+				s2 := &http.Server{}
+
+				go s1.Serve(l1)
+				go s2.Serve(l2)
+
+				// Make sure they're up
+				_, err = http.Get("http://" + l1.Addr().String())
+				require.NoError(t, err)
+
+				c := http.Client{Transport: &http.Transport{
+					Dial: func(network, addr string) (net.Conn, error) {
+						return net.Dial("unix", l2.Addr().String())
+					},
+				}}
+				_, err = c.Get("http://random.domain")
+				require.NoError(t, err)
+
+				return &Worker{
+					baseContext: context.Background(),
+					listeners: []*base.ServerListener{
+						{
+							ALPNListener: l1,
+							HTTPServer:   s1,
+							Mux:          alpnmux.New(l1),
+							Config:       &listenerutil.ListenerConfig{Type: "tcp"},
+						},
+						{
+							ALPNListener: l2,
+							HTTPServer:   s2,
+							Mux:          alpnmux.New(l2),
+							Config:       &listenerutil.ListenerConfig{Type: "tcp"},
+						},
+					},
+				}
+			},
+			assertions: func(t *testing.T, w *Worker) {
+				// Asserts the HTTP Servers are closed.
+				require.ErrorIs(t, w.listeners[0].HTTPServer.Serve(w.listeners[0].ALPNListener), http.ErrServerClosed)
+				require.ErrorIs(t, w.listeners[1].HTTPServer.Serve(w.listeners[1].ALPNListener), http.ErrServerClosed)
+
+				// Asserts the underlying listeners are closed.
+				require.ErrorIs(t, w.listeners[0].Mux.Close(), net.ErrClosed)
+				require.ErrorIs(t, w.listeners[1].Mux.Close(), net.ErrClosed)
+			},
+			expErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			w := tt.workerFn(t)
+			err := w.stopHttpServersAndListeners()
+			if tt.expErr {
+				require.EqualError(t, err, tt.expErrStr)
+				return
+			}
+
+			require.NoError(t, err)
+			if tt.assertions != nil {
+				tt.assertions(t, w)
+			}
+		})
+	}
+}
+
+func TestStopAnyListeners(t *testing.T) {
+	tests := []struct {
+		name       string
+		workerFn   func(t *testing.T) *Worker
+		assertions func(t *testing.T, w *Worker)
+		expErr     bool
+		expErrStr  string
+	}{
+		{
+			name: "nil listeners",
+			workerFn: func(t *testing.T) *Worker {
+				return &Worker{listeners: nil}
+			},
+			expErr: false,
+		},
+		{
+			name: "no listeners",
+			workerFn: func(t *testing.T) *Worker {
+				return &Worker{listeners: []*base.ServerListener{}}
+			},
+			expErr: false,
+		},
+		{
+			name: "listeners slice with nil server listeners",
+			workerFn: func(t *testing.T) *Worker {
+				return &Worker{listeners: []*base.ServerListener{nil, nil, nil}}
+			},
+			expErr: false,
+		},
+		{
+			name: "listeners with nil mux",
+			workerFn: func(t *testing.T) *Worker {
+				return &Worker{listeners: []*base.ServerListener{
+					{Mux: nil}, {Mux: nil}, {Mux: nil},
+				}}
+			},
+			expErr: false,
+		},
+		{
+			name: "multiple listeners, including a closed one",
+			workerFn: func(t *testing.T) *Worker {
+				l1, err := net.Listen("tcp", "127.0.0.1:0")
+				require.NoError(t, err)
+
+				l2, err := net.Listen("unix", "/tmp/boundary-worker-TestStopAnyListeners-"+strconv.FormatInt(time.Now().UnixNano(), 10))
+				require.NoError(t, err)
+
+				l3, err := net.Listen("tcp", "127.0.0.1:0")
+				require.NoError(t, err)
+				require.NoError(t, l3.Close())
+
+				return &Worker{listeners: []*base.ServerListener{
+					{
+						Config: &listenerutil.ListenerConfig{Type: "tcp"},
+						Mux:    alpnmux.New(l1),
+					},
+					{
+						Config: &listenerutil.ListenerConfig{Type: "tcp"},
+						Mux:    alpnmux.New(l2),
+					},
+					{
+						Config: &listenerutil.ListenerConfig{Type: "tcp"},
+						Mux:    alpnmux.New(l3),
+					},
+				}}
+			},
+			assertions: func(t *testing.T, w *Worker) {
+				for i := range w.listeners {
+					ln := w.listeners[i]
+					require.ErrorIs(t, ln.Mux.Close(), net.ErrClosed)
+				}
+			},
+			expErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := tt.workerFn(t)
+			err := c.stopAnyListeners()
+			if tt.expErr {
+				require.EqualError(t, err, tt.expErrStr)
+				return
+			}
+
+			require.NoError(t, err)
+			if tt.assertions != nil {
+				tt.assertions(t, c)
+			}
+		})
+	}
+}
+
+func TestListenerCloseErrorCheck(t *testing.T) {
+	tests := []struct {
+		name   string
+		lnType string
+		err    error
+		expErr error
+	}{
+		{
+			name:   "nil err",
+			lnType: "tcp",
+			err:    nil,
+			expErr: nil,
+		},
+		{
+			name:   "net.Closed",
+			lnType: "tcp",
+			err:    net.ErrClosed,
+			expErr: nil,
+		},
+		{
+			name:   "path err not unix type",
+			lnType: "tcp",
+			err:    &os.PathError{Op: "test"},
+			expErr: &os.PathError{Op: "test"},
+		},
+		{
+			name:   "path err unix type",
+			lnType: "unix",
+			err:    &os.PathError{Op: "test"},
+			expErr: nil,
+		},
+		{
+			name:   "literally anything else",
+			lnType: "tcp",
+			err:    fmt.Errorf("oops I errored"),
+			expErr: fmt.Errorf("oops I errored"),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := listenerCloseErrorCheck(tt.lnType, tt.err)
+			require.Equal(t, tt.expErr, err)
+		})
+	}
+}

internal/servers/worker/testing.go

@@ -148,7 +148,7 @@ func (tw *TestWorker) Shutdown() {
 	tw.cancel()
 
 	if tw.w != nil {
-		if err := tw.w.Shutdown(false); err != nil {
+		if err := tw.w.Shutdown(); err != nil {
 			tw.t.Error(err)
 		}
 	}
@@ -183,8 +183,9 @@ type TestWorkerOpts struct {
 	// connection cannot be made back to the controller
 	StatusGracePeriodDuration time.Duration
 
-	// Whether to set the replay value
-	EnableAuthReplay bool
+	// Overrides worker's nonceFn, for cases where we want to have control
+	// over the nonce we send to the Controller
+	NonceFn randFn
 }
 
 func NewTestWorker(t *testing.T, opts *TestWorkerOpts) *TestWorker {
@@ -280,8 +281,8 @@ func NewTestWorker(t *testing.T, opts *TestWorkerOpts) *TestWorker {
 		t.Fatal(err)
 	}
 
-	if opts.EnableAuthReplay {
-		tw.w.testReuseAuthNonces = true
+	if opts.NonceFn != nil {
+		tw.w.nonceFn = opts.NonceFn
 	}
 
 	if !opts.DisableAutoStart {

internal/servers/worker/worker.go

@@ -14,18 +14,22 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/servers"
 	"github.com/hashicorp/boundary/internal/servers/worker/session"
 	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
 	ua "go.uber.org/atomic"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 )
 
+type randFn func(length int) (string, error)
+
 type Worker struct {
 	conf   *Config
 	logger hclog.Logger
@@ -45,6 +49,11 @@ type Worker struct {
 	controllerSessionConn *atomic.Value
 	sessionInfoMap        *sync.Map
 
+	listeners []*base.ServerListener
+
+	// Used to generate a random nonce for Controller connections
+	nonceFn randFn
+
 	// We store the current set in an atomic value so that we can add
 	// reload-on-sighup behavior later
 	tags *atomic.Value
@@ -52,10 +61,6 @@ type Worker struct {
 	// request. It can be set via startup in New below, or (eventually) via
 	// SIGHUP.
 	updateTags ua.Bool
-
-	// Test-related values
-	testReuseAuthNonces bool
-	testReusedAuthNonce string
 }
 
 func New(conf *Config) (*Worker, error) {
@@ -69,6 +74,7 @@ func New(conf *Config) (*Worker, error) {
 		controllerSessionConn: new(atomic.Value),
 		sessionInfoMap:        new(sync.Map),
 		tags:                  new(atomic.Value),
+		nonceFn:               base62.Random,
 	}
 
 	w.lastStatusSuccess.Store((*LastStatusInformation)(nil))
@@ -105,6 +111,23 @@ func New(conf *Config) (*Worker, error) {
 		}
 	}
 
+	for i := range conf.Listeners {
+		l := conf.Listeners[i]
+		if l == nil || l.Config == nil || l.Config.Purpose == nil {
+			continue
+		}
+		if len(l.Config.Purpose) != 1 {
+			return nil, fmt.Errorf("found listener with multiple purposes %q", strings.Join(l.Config.Purpose, ","))
+		}
+		switch l.Config.Purpose[0] {
+		case "proxy":
+			w.listeners = append(w.listeners, l)
+		}
+	}
+	if len(w.listeners) == 0 {
+		return nil, fmt.Errorf("no proxy listeners found")
+	}
+
 	return w, nil
 }
 
@@ -144,7 +167,7 @@ func (w *Worker) Start() error {
 // listeners, useful for tests if we want to stop and start a worker. In order
 // to create new listeners we'd have to migrate listener setup logic here --
 // doable, but work for later.
-func (w *Worker) Shutdown(skipListeners bool) error {
+func (w *Worker) Shutdown() error {
 	const op = "worker.(Worker).Shutdown"
 	if !w.started.Load() {
 		event.WriteSysEvent(w.baseContext, op, "already shut down, skipping")
@@ -156,10 +179,9 @@ func (w *Worker) Shutdown(skipListeners bool) error {
 	defer w.started.Store(false)
 	w.Resolver().UpdateState(resolver.State{Addresses: []resolver.Address{}})
 	w.baseCancel()
-	if !skipListeners {
-		if err := w.stopListeners(); err != nil {
-			return fmt.Errorf("error stopping worker listeners: %w", err)
-		}
+
+	if err := w.stopServersAndListeners(); err != nil {
+		return fmt.Errorf("error stopping worker servers and listeners: %w", err)
 	}
 
 	// Shut down all connections.

internal/servers/worker/worker_test.go

@@ -0,0 +1,114 @@
+package worker
+
+import (
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/cmd/base"
+	"github.com/hashicorp/boundary/internal/cmd/config"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-secure-stdlib/configutil/v2"
+	"github.com/hashicorp/go-secure-stdlib/listenerutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerNew(t *testing.T) {
+	tests := []struct {
+		name       string
+		in         *Config
+		expErr     bool
+		expErrMsg  string
+		assertions func(t *testing.T, w *Worker)
+	}{
+		{
+			name:      "nil listeners",
+			in:        &Config{Server: &base.Server{Listeners: nil}},
+			expErr:    true,
+			expErrMsg: "no proxy listeners found",
+		},
+		{
+			name:      "zero listeners",
+			in:        &Config{Server: &base.Server{Listeners: []*base.ServerListener{}}},
+			expErr:    true,
+			expErrMsg: "no proxy listeners found",
+		},
+		{
+			name: "populated with nil values",
+			in: &Config{
+				Server: &base.Server{
+					Listeners: []*base.ServerListener{
+						nil,
+						{Config: nil},
+						{Config: &listenerutil.ListenerConfig{Purpose: nil}},
+					},
+				},
+			},
+			expErr:    true,
+			expErrMsg: "no proxy listeners found",
+		},
+		{
+			name: "multiple purposes",
+			in: &Config{
+				Server: &base.Server{
+					Listeners: []*base.ServerListener{
+						nil,
+						{Config: nil},
+						{Config: &listenerutil.ListenerConfig{Purpose: []string{"api", "proxy"}}},
+					},
+				},
+			},
+			expErr:    true,
+			expErrMsg: `found listener with multiple purposes "api,proxy"`,
+		},
+		{
+			name: "valid listeners",
+			in: &Config{
+				Server: &base.Server{
+					Listeners: []*base.ServerListener{
+						{Config: &listenerutil.ListenerConfig{Purpose: []string{"api"}}},
+						{Config: &listenerutil.ListenerConfig{Purpose: []string{"proxy"}}},
+						{Config: &listenerutil.ListenerConfig{Purpose: []string{"proxy"}}},
+						{Config: &listenerutil.ListenerConfig{Purpose: []string{"cluster"}}},
+					},
+				},
+			},
+			expErr: false,
+			assertions: func(t *testing.T, w *Worker) {
+				require.Len(t, w.listeners, 2)
+			},
+		},
+		{
+			name: "worker nonce func is set",
+			in: &Config{
+				Server: &base.Server{
+					Listeners: []*base.ServerListener{
+						{Config: &listenerutil.ListenerConfig{Purpose: []string{"proxy"}}},
+					},
+				},
+			},
+			expErr: false,
+			assertions: func(t *testing.T, w *Worker) {
+				require.NotNil(t, w.nonceFn)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// New() panics if these aren't set
+			tt.in.Logger = hclog.Default()
+			tt.in.RawConfig = &config.Config{SharedConfig: &configutil.SharedConfig{DisableMlock: true}}
+
+			w, err := New(tt.in)
+			if tt.expErr {
+				require.EqualError(t, err, tt.expErrMsg)
+				require.Nil(t, w)
+				return
+			}
+
+			require.NoError(t, err)
+			if tt.assertions != nil {
+				tt.assertions(t, w)
+			}
+		})
+	}
+}

internal/tests/cluster/ipv6_listener_test.go

@@ -11,12 +11,11 @@ import (
 	"github.com/hashicorp/boundary/internal/servers/controller"
 	"github.com/hashicorp/boundary/internal/servers/worker"
 	"github.com/hashicorp/go-hclog"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func TestIPv6Listener(t *testing.T) {
-	assert, require := assert.New(t), require.New(t)
+	require := require.New(t)
 	logger := hclog.New(&hclog.LoggerOptions{
 		Level: hclog.Trace,
 	})
@@ -40,28 +39,7 @@ func TestIPv6Listener(t *testing.T) {
 	})
 	defer c1.Shutdown()
 
-	expectWorkers := func(c *controller.TestController, workers ...*worker.TestWorker) {
-		updateTimes := c.Controller().WorkerStatusUpdateTimes()
-		workerMap := map[string]*worker.TestWorker{}
-		for _, w := range workers {
-			workerMap[w.Name()] = w
-		}
-		updateTimes.Range(func(k, v interface{}) bool {
-			require.NotNil(k)
-			require.NotNil(v)
-			if workerMap[k.(string)] == nil {
-				// We don't remove from updateTimes currently so if we're not
-				// expecting it we'll see an out-of-date entry
-				return true
-			}
-			assert.WithinDuration(time.Now(), v.(time.Time), 60*time.Second)
-			delete(workerMap, k.(string))
-			return true
-		})
-		assert.Empty(workerMap)
-	}
-
-	expectWorkers(c1)
+	expectWorkers(t, c1)
 
 	wconf, err := config.DevWorker()
 	require.NoError(err)
@@ -75,7 +53,7 @@ func TestIPv6Listener(t *testing.T) {
 	defer w1.Shutdown()
 
 	time.Sleep(10 * time.Second)
-	expectWorkers(c1, w1)
+	expectWorkers(t, c1, w1)
 
 	c2 := c1.AddClusterControllerMember(t, &controller.TestControllerOpts{
 		Logger: c1.Config().Logger.ResetNamed("c2"),
@@ -83,12 +61,12 @@ func TestIPv6Listener(t *testing.T) {
 	defer c2.Shutdown()
 
 	time.Sleep(10 * time.Second)
-	expectWorkers(c2, w1)
+	expectWorkers(t, c2, w1)
 
-	require.NoError(w1.Worker().Shutdown(true))
+	require.NoError(w1.Worker().Shutdown())
 	time.Sleep(10 * time.Second)
-	expectWorkers(c1)
-	expectWorkers(c2)
+	expectWorkers(t, c1)
+	expectWorkers(t, c2)
 
 	client, err := api.NewClient(nil)
 	require.NoError(err)

internal/tests/cluster/multi_controller_worker_test.go

@@ -54,21 +54,31 @@ func TestMultiControllerMultiWorkerConnections(t *testing.T) {
 	expectWorkers(t, c1, w1, w2)
 	expectWorkers(t, c2, w1, w2)
 
-	require.NoError(w1.Worker().Shutdown(true))
+	require.NoError(w1.Worker().Shutdown())
 	time.Sleep(10 * time.Second)
 	expectWorkers(t, c1, w2)
 	expectWorkers(t, c2, w2)
 
-	require.NoError(w1.Worker().Start())
+	w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{
+		WorkerAuthKms:      c1.Config().WorkerAuthKms,
+		InitialControllers: c1.ClusterAddrs(),
+		Logger:             logger.Named("w1"),
+	})
+	defer w1.Shutdown()
+
 	time.Sleep(10 * time.Second)
 	expectWorkers(t, c1, w1, w2)
 	expectWorkers(t, c2, w1, w2)
 
 	require.NoError(c2.Controller().Shutdown())
 	time.Sleep(10 * time.Second)
-	expectWorkers(t, c2, w1, w2)
+	expectWorkers(t, c1, w1, w2)
+
+	c2 = c1.AddClusterControllerMember(t, &controller.TestControllerOpts{
+		Logger: c1.Config().Logger.ResetNamed("c2"),
+	})
+	defer c2.Shutdown()
 
-	require.NoError(c1.Controller().Start())
 	time.Sleep(10 * time.Second)
 	expectWorkers(t, c1, w1, w2)
 	expectWorkers(t, c2, w1, w2)

internal/tests/cluster/unix_listener_test.go

@@ -14,12 +14,11 @@ import (
 	"github.com/hashicorp/boundary/internal/servers/controller"
 	"github.com/hashicorp/boundary/internal/servers/worker"
 	"github.com/hashicorp/go-hclog"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func TestUnixListener(t *testing.T) {
-	assert, require := assert.New(t), require.New(t)
+	require := require.New(t)
 	logger := hclog.New(&hclog.LoggerOptions{
 		Level: hclog.Trace,
 	})
@@ -53,28 +52,7 @@ func TestUnixListener(t *testing.T) {
 	})
 	defer c1.Shutdown()
 
-	expectWorkers := func(c *controller.TestController, workers ...*worker.TestWorker) {
-		updateTimes := c.Controller().WorkerStatusUpdateTimes()
-		workerMap := map[string]*worker.TestWorker{}
-		for _, w := range workers {
-			workerMap[w.Name()] = w
-		}
-		updateTimes.Range(func(k, v interface{}) bool {
-			require.NotNil(k)
-			require.NotNil(v)
-			if workerMap[k.(string)] == nil {
-				// We don't remove from updateTimes currently so if we're not
-				// expecting it we'll see an out-of-date entry
-				return true
-			}
-			assert.WithinDuration(time.Now(), v.(time.Time), 60*time.Second)
-			delete(workerMap, k.(string))
-			return true
-		})
-		assert.Empty(workerMap)
-	}
-
-	expectWorkers(c1)
+	expectWorkers(t, c1)
 
 	wconf, err := config.DevWorker()
 	require.NoError(err)
@@ -88,11 +66,11 @@ func TestUnixListener(t *testing.T) {
 	defer w1.Shutdown()
 
 	time.Sleep(10 * time.Second)
-	expectWorkers(c1, w1)
+	expectWorkers(t, c1, w1)
 
-	require.NoError(w1.Worker().Shutdown(true))
+	require.NoError(w1.Worker().Shutdown())
 	time.Sleep(10 * time.Second)
-	expectWorkers(c1)
+	expectWorkers(t, c1)
 
 	require.NoError(c1.Controller().Shutdown())
 	c1 = controller.NewTestController(t, &controller.TestControllerOpts{
@@ -103,7 +81,7 @@ func TestUnixListener(t *testing.T) {
 	defer c1.Shutdown()
 
 	time.Sleep(10 * time.Second)
-	expectWorkers(c1)
+	expectWorkers(t, c1)
 
 	client, err := api.NewClient(nil)
 	require.NoError(err)

internal/tests/cluster/worker_replay_test.go

@@ -13,7 +13,6 @@ import (
 	"github.com/hashicorp/boundary/internal/servers/controller"
 	"github.com/hashicorp/boundary/internal/servers/worker"
 	"github.com/hashicorp/go-hclog"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -41,24 +40,49 @@ func TestWorkerReplay(t *testing.T) {
 		Config:             conf,
 		WorkerAuthKms:      c1.Config().WorkerAuthKms,
 		InitialControllers: c1.ClusterAddrs(),
-		EnableAuthReplay:   true,
+		NonceFn: func(length int) (string, error) {
+			return "test_noncetest_nonce", nil
+		},
 	})
 
 	// Give time for it to connect
 	time.Sleep(10 * time.Second)
-	require.NoError(t, w1.Worker().Shutdown(true))
 
-	// Now, start up again
-	require.NoError(t, w1.Worker().Start())
+	// Assert that the worker has connected
+	logBuf, err := os.ReadFile(ec.AllEvents.Name())
+	require.NoError(t, err)
+	require.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed"))
+
+	// Assert we have the expected nonce in the DB
+	nonces, err := c1.ServersRepo().ListNonces(c1.Context(), servers.NoncePurposeWorkerAuth)
+	require.NoError(t, err)
+	require.Len(t, nonces, 1)
+	require.Equal(t, &servers.Nonce{Nonce: "test_noncetest_nonce", Purpose: servers.NoncePurposeWorkerAuth}, nonces[0])
+
+	require.NoError(t, w1.Worker().Shutdown())
+
+	// Now, start up again with the same nonce
+	w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{
+		Config:             conf,
+		WorkerAuthKms:      c1.Config().WorkerAuthKms,
+		InitialControllers: c1.ClusterAddrs(),
+		NonceFn: func(length int) (string, error) {
+			return "test_noncetest_nonce", nil
+		},
+	})
+
+	// Give time for it to connect
 	time.Sleep(10 * time.Second)
 
 	// We should find only one nonce, and one successful worker authentication,
 	// both in the output and in the database
 	ec.AllEvents.Close()
-	logBuf, err := os.ReadFile(ec.AllEvents.Name())
+	logBuf, err = os.ReadFile(ec.AllEvents.Name())
 	require.NoError(t, err)
-	assert.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed"))
-	nonces, err := c1.ServersRepo().ListNonces(c1.Context(), servers.NoncePurposeWorkerAuth)
+	require.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed"))
+
+	nonces, err = c1.ServersRepo().ListNonces(c1.Context(), servers.NoncePurposeWorkerAuth)
 	require.NoError(t, err)
-	assert.Len(t, nonces, 1)
+	require.Len(t, nonces, 1)
+	require.Equal(t, &servers.Nonce{Nonce: "test_noncetest_nonce", Purpose: servers.NoncePurposeWorkerAuth}, nonces[0])
 }