initial commit (#2768)

go.mod

@@ -94,7 +94,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.1.3
 	github.com/hashicorp/go-kms-wrapping/extras/kms/v2 v2.0.0-20221122211539-47c893099f13
 	github.com/hashicorp/go-version v1.3.0
-	github.com/hashicorp/nodeenrollment v0.1.17
+	github.com/hashicorp/nodeenrollment v0.1.18
 	github.com/kelseyhightower/envconfig v1.4.0
 	golang.org/x/exp v0.0.0-20220921164117-439092de6870
 	golang.org/x/net v0.0.0-20220722155237-a158d28d115b
@@ -139,7 +139,7 @@ require (
 	github.com/hashicorp/go-secure-stdlib/tlsutil v0.1.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/hashicorp/vault/sdk v0.3.0 // indirect
-	github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d // indirect
+	github.com/hashicorp/yamux v0.1.0 // indirect
 	github.com/huandu/xstrings v1.3.2 // indirect
 	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/jackc/chunkreader/v2 v2.0.1 // indirect

go.sum

@@ -744,16 +744,17 @@ github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+l
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/hashicorp/nodeenrollment v0.1.17 h1:ZaGNugd3EOZIdJxgC5bUA1CbHZ/OKZcgXnquzoFql6E=
-github.com/hashicorp/nodeenrollment v0.1.17/go.mod h1:N5gYsm8mWiDfIw/j+1IQ6NBO1cWCmhPpvQ9GB1QUnsU=
+github.com/hashicorp/nodeenrollment v0.1.18 h1:fqFMZ5e2cWupeQDxKledF9DGHsk9GoJl7K2Wrdiys6k=
+github.com/hashicorp/nodeenrollment v0.1.18/go.mod h1:N5gYsm8mWiDfIw/j+1IQ6NBO1cWCmhPpvQ9GB1QUnsU=
 github.com/hashicorp/vault/api v1.3.1 h1:pkDkcgTh47PRjY1NEFeofqR4W/HkNUi9qIakESO2aRM=
 github.com/hashicorp/vault/api v1.3.1/go.mod h1:QeJoWxMFt+MsuWcYhmwRLwKEXrjwAFFywzhptMsTIUw=
 github.com/hashicorp/vault/sdk v0.1.13/go.mod h1:B+hVj7TpuQY1Y/GPbCpffmgd+tSEwvhkWnjtSYCaS2M=
 github.com/hashicorp/vault/sdk v0.3.0 h1:kR3dpxNkhh/wr6ycaJYqp6AFT/i2xaftbfnwZduTKEY=
 github.com/hashicorp/vault/sdk v0.3.0/go.mod h1:aZ3fNuL5VNydQk8GcLJ2TV8YCRVvyaakYkhZRoVuhj0=
 github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM=
-github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d h1:kJCB4vdITiW1eC1vq2e6IsrXKrZit1bv/TDYFGMp4BQ=
 github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM=
+github.com/hashicorp/yamux v0.1.0 h1:DzDIF6Sd7GD2sX0kDFpHAsJMY4L+OfTvtuaQsOYXxzk=
+github.com/hashicorp/yamux v0.1.0/go.mod h1:CtWFDAQgb7dxtzFs4tWbplKIe2jSi3+5vKbgIO0SLnQ=
 github.com/hinshun/vt10x v0.0.0-20180616224451-1954e6464174 h1:WlZsjVhE8Af9IcZDGgJGQpNflI3+MJSBhsgT5PCtzBQ=
 github.com/hinshun/vt10x v0.0.0-20180616224451-1954e6464174/go.mod h1:DqJ97dSdRW1W22yXSB90986pcOyQ7r45iio1KN2ez1A=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=

internal/daemon/common/const.go

@@ -1,3 +1,6 @@
 package common
 
-const ReverseGrpcConnectionAlpnValue = "the-downstream-dialer-plays-an-uno-reverse-card"
+const (
+	ReverseGrpcConnectionAlpnValue = "the-downstream-dialer-plays-an-uno-reverse-card"
+	DataPlaneProxyAlpnValue        = "i-herd-you-like-proxies-so-i-put-a-proxy-in-your-proxy"
+)

internal/daemon/controller/controller.go

@@ -39,6 +39,7 @@ import (
 	host_plugin_assets "github.com/hashicorp/boundary/plugins/host"
 	"github.com/hashicorp/boundary/sdk/pbs/plugin"
 	external_host_plugins "github.com/hashicorp/boundary/sdk/plugins/host"
+	"github.com/hashicorp/boundary/version"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
 	"github.com/hashicorp/go-secure-stdlib/pluginutil/v2"
@@ -56,7 +57,7 @@ type downstreamRouter interface {
 	// ProcessPendingConnections starts a function that continually processes
 	// incoming client connections. This only returns when the provided context
 	// is done.
-	StartProcessingPendingConnections(context.Context, func() string)
+	StartProcessingPendingConnections(context.Context, func() string) error
 }
 
 // downstreamWorkersTicker defines an interface for a ticker that maintains the
@@ -70,8 +71,8 @@ type downstreamWorkersTicker interface {
 var (
 	downstreamRouterFactory func() downstreamRouter
 
-	downstreamersFactory           func(context.Context, string) (common.Downstreamers, error)
-	downstreamWorkersTickerFactory func(context.Context, string, common.Downstreamers, downstreamRouter) (downstreamWorkersTicker, error)
+	downstreamersFactory           func(context.Context, string, string) (common.Downstreamers, error)
+	downstreamWorkersTickerFactory func(context.Context, string, string, common.Downstreamers, downstreamRouter) (downstreamWorkersTicker, error)
 	commandClientFactory           func(context.Context, *Controller) error
 )
 
@@ -389,7 +390,8 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 	}
 
 	if downstreamersFactory != nil {
-		c.downstreamWorkers, err = downstreamersFactory(ctx, "root")
+		boundVer := version.Get().VersionNumber()
+		c.downstreamWorkers, err = downstreamersFactory(ctx, "root", boundVer)
 		if err != nil {
 			return nil, fmt.Errorf("unable to initialize downstream workers graph: %w", err)
 		}
@@ -479,7 +481,8 @@ func (c *Controller) Start() error {
 	if downstreamWorkersTickerFactory != nil {
 		// we'll use "root" to designate that this is the root of the graph (aka
 		// a controller)
-		dswTicker, err := downstreamWorkersTickerFactory(c.baseContext, "root", c.downstreamWorkers, c.downstreamRoutes)
+		boundVer := version.Get().VersionNumber()
+		dswTicker, err := downstreamWorkersTickerFactory(c.baseContext, "root", boundVer, c.downstreamWorkers, c.downstreamRoutes)
 		if err != nil {
 			return fmt.Errorf("error creating downstream workers ticker: %w", err)
 		}

internal/daemon/controller/listeners.go

@@ -26,7 +26,7 @@ import (
 // the function that handles a secondary connection over a provided listener
 var handleSecondaryConnection = closeListener
 
-func closeListener(_ context.Context, l net.Listener, _ any, _ int) error {
+func closeListener(_ context.Context, l, _ net.Listener, _ any) error {
 	if l != nil {
 		return l.Close()
 	}
@@ -243,18 +243,17 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 	ln.GrpcServer = workerServer
 
 	return func() {
+		err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(multiplexingReverseGrpcListener, "reverse-grpc"),
+			nil, c.downstreamRoutes)
+		if err != nil {
+			event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error"))
+		}
 		go func() {
 			err := splitListener.Start()
 			if err != nil {
 				event.WriteError(c.baseContext, op, err, event.WithInfoMsg("splitListener.Start() error"))
 			}
 		}()
-		go func() {
-			err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(multiplexingReverseGrpcListener, "reverse-grpc"), c.downstreamRoutes, -1)
-			if err != nil {
-				event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error"))
-			}
-		}()
 		go func() {
 			err := ln.GrpcServer.Serve(metric.InstrumentClusterTrackingListener(multiplexingAuthedListener, "grpc"))
 			if err != nil {

internal/daemon/worker/addressreceiver.go

@@ -16,9 +16,9 @@ func noopAddressReceivers(context.Context, *Worker) ([]addressReceiver, error) {
 type receiverType uint
 
 const (
-	UnknownReceiverType         receiverType = 0
-	grpcResolverReceiverType    receiverType = 1
-	dialingListenerReceiverType receiverType = 2
+	UnknownReceiverType receiverType = iota
+	grpcResolverReceiverType
+	secondaryConnectionReceiverType
 )
 
 // String returns a string representation of the receiverType
@@ -26,7 +26,7 @@ func (s receiverType) String() string {
 	return [...]string{
 		"unknown",
 		"grpcResolver",
-		"dialingListener",
+		"secondaryConnections",
 	}[s]
 }
 

internal/daemon/worker/controller_connection.go

@@ -33,6 +33,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 const hcpbUrlSuffix = ".proxy.boundary.hashicorp.cloud:9202"
@@ -77,8 +78,14 @@ func (w *Worker) StartControllerConnections() error {
 	return nil
 }
 
-func (w *Worker) controllerDialerFunc(extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) {
-	const op = "worker.(Worker).controllerDialerFunc"
+// upstreamDialerFunc dials an upstream server. extraAlpnProtos can be provided
+// to kms and pki connections and are used for identifying, on the server side,
+// the intended purpose of the connection.  upstreamDialerFunc takes an optional
+// state struct which, if the connection is made using the node enrolment
+// library, sends the state to the server and can be retrieved from the
+// resulting protocol.Conn.
+func (w *Worker) upstreamDialerFunc(state *structpb.Struct, extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) {
+	const op = "worker.(Worker).upstreamDialerFunc"
 	return func(ctx context.Context, addr string) (net.Conn, error) {
 		var conn net.Conn
 		var err error
@@ -90,6 +97,7 @@ func (w *Worker) controllerDialerFunc(extraAlpnProtos ...string) func(context.Co
 				ctx,
 				w.WorkerAuthStorage,
 				addr,
+				nodeenrollment.WithState(state),
 				nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms),
 				nodeenrollment.WithExtraAlpnProtos(extraAlpnProtos),
 				// If the activation token hasn't been populated, this won't do
@@ -202,7 +210,7 @@ func (w *Worker) createClientConn(addr string) error {
 		grpc.WithUnaryInterceptor(metric.InstrumentClusterClient()),
 		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(math.MaxInt32)),
 		grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(math.MaxInt32)),
-		grpc.WithContextDialer(w.controllerDialerFunc()),
+		grpc.WithContextDialer(w.upstreamDialerFunc(nil)),
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithDefaultServiceConfig(defServiceConfig),
 		// Don't have the resolver reach out for a service config from the

internal/daemon/worker/handler.go

@@ -196,9 +196,9 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 		}
 		workerId := w.LastStatusSuccess().WorkerId
 
-		var ci *pbs.AuthorizeConnectionResponse
+		var acResp *pbs.AuthorizeConnectionResponse
 		var connsLeft int32
-		ci, connsLeft, err = sess.RequestAuthorizeConnection(ctx, workerId, connCancel)
+		acResp, connsLeft, err = sess.RequestAuthorizeConnection(ctx, workerId, connCancel)
 		if err != nil {
 			event.WriteError(ctx, op, err, event.WithInfoMsg("unable to authorize connection"))
 			if err = conn.Close(websocket.StatusInternalError, "unable to authorize connection"); err != nil {
@@ -206,12 +206,12 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 			}
 			return
 		}
-		event.WriteSysEvent(ctx, op, "connection successfully authorized", "session_id", sessionId, "connection_id", ci.GetConnectionId())
+		event.WriteSysEvent(ctx, op, "connection successfully authorized", "session_id", sessionId, "connection_id", acResp.GetConnectionId())
 
 		// Wrapping the client websocket with a `net.Conn` implementation that
 		// records the bytes that go across Read() and Write().
 		cc := &countingConn{Conn: websocket.NetConn(connCtx, conn, websocket.MessageBinary)}
-		err = sess.ApplyConnectionCounterCallbacks(ci.GetConnectionId(), cc.BytesRead, cc.BytesWritten)
+		err = sess.ApplyConnectionCounterCallbacks(acResp.GetConnectionId(), cc.BytesRead, cc.BytesWritten)
 		if err != nil {
 			event.WriteError(ctx, op, err, event.WithInfoMsg("unable to set counter callbacks for session connection"))
 			err = conn.Close(websocket.StatusInternalError, "unable to set counter callbacks for session connection")
@@ -223,14 +223,14 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 
 		defer func() {
 			ccd := map[string]*session.ConnectionCloseData{
-				ci.GetConnectionId(): {
+				acResp.GetConnectionId(): {
 					SessionId: sess.GetId(),
 					BytesUp:   cc.BytesRead(),
 					BytesDown: cc.BytesWritten(),
 				},
 			}
 			if sessionManager.RequestCloseConnections(ctx, ccd) {
-				event.WriteSysEvent(ctx, op, "connection closed", "session_id", sessionId, "connection_id", ci.GetConnectionId())
+				event.WriteSysEvent(ctx, op, "connection closed", "session_id", sessionId, "connection_id", acResp.GetConnectionId())
 			}
 		}()
 
@@ -255,7 +255,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 			}
 			return
 		}
-		protocolCtx := ci.GetProtocolContext()
+		protocolCtx := acResp.GetProtocolContext()
 		if protocolCtx == nil {
 			// TODO: Remove this if block once pre v0.12.0 controllers are no longer supported.
 			if protocolCtx, err = GetProtocolContext(ctx, workerId, sess, endpointUrl.Scheme); err != nil {
@@ -265,7 +265,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 			}
 		}
 
-		pDialer, err := proxyHandlers.GetEndpointDialer(ctx, endpointUrl.Host, workerId, ci, w.downstreamRoutes)
+		pDialer, err := proxyHandlers.GetEndpointDialer(ctx, endpointUrl.Host, workerId, acResp, w.downstreamRoutes)
 		if err != nil {
 			conn.Close(proxyHandlers.WebsocketStatusProtocolSetupError, "unable to get endpoint dialer")
 			event.WriteError(ctx, op, err)
@@ -273,13 +273,13 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 		}
 
 		// Verify the protocol has a supported proxy before calling RequestAuthorizeConnection
-		handleProxyFn, err := proxyHandlers.GetHandler(workerId, ci.GetProtocolContext())
+		handleProxyFn, err := proxyHandlers.GetHandler(workerId, acResp.GetProtocolContext())
 		if err != nil {
 			conn.Close(proxyHandlers.WebsocketStatusProtocolSetupError, "unable to get proxy handler")
 			event.WriteError(ctx, op, err)
 			return
 		}
-		runProxy, err := handleProxyFn(ctx, cc, pDialer, ci.GetConnectionId(), protocolCtx)
+		runProxy, err := handleProxyFn(ctx, cc, pDialer, acResp.GetConnectionId(), protocolCtx)
 		if err != nil {
 			conn.Close(proxyHandlers.WebsocketStatusProtocolSetupError, "unable to setup proxying")
 			event.WriteError(ctx, op, err)
@@ -290,7 +290,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 		// that we can establish the proxy.
 		endpointAddr := pDialer.LastConnectionAddr()
 		connectionInfo := &pbs.ConnectConnectionRequest{
-			ConnectionId:       ci.GetConnectionId(),
+			ConnectionId:       acResp.GetConnectionId(),
 			ClientTcpAddress:   clientAddr.IP.String(),
 			ClientTcpPort:      uint32(clientAddr.Port),
 			EndpointTcpAddress: endpointAddr.Ip(),
@@ -299,7 +299,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 			UserClientIp:       userClientIp,
 		}
 		if err = sess.RequestConnectConnection(ctx, connectionInfo); err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error requesting connect connection", "session_id", sess.GetId(), "connection_id", ci.GetConnectionId()))
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error requesting connect connection", "session_id", sess.GetId(), "connection_id", acResp.GetConnectionId()))
 			if err = conn.Close(websocket.StatusInternalError, "unable to establish proxy"); err != nil {
 				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
 			}

internal/daemon/worker/listeners.go

@@ -30,11 +30,16 @@ import (
 )
 
 // the function that handles a secondary connection over a provided listener
-var handleSecondaryConnection = closeListener
+var handleSecondaryConnection = closeListeners
 
-func closeListener(_ context.Context, l net.Listener, _ any, _ int) error {
+// closeListeners handles the secondary connection listeners by closing them.
+// l is the grpc listener and l2 is the data plane listener.
+func closeListeners(_ context.Context, l, l2 net.Listener, _ any) error {
 	if l != nil {
-		return l.Close()
+		l.Close()
+	}
+	if l2 != nil {
+		l2.Close()
 	}
 	return nil
 }
@@ -168,9 +173,9 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 
 	// Connections coming in here are authed by nodeenrollment and are for the
 	// reverse grpc purpose
-	reverseGrpcListener, err := w.workerAuthSplitListener.GetListener(common.ReverseGrpcConnectionAlpnValue)
+	reverseGrpcListener, err := w.workerAuthSplitListener.GetListener(common.ReverseGrpcConnectionAlpnValue, nodee.WithNativeConns(true))
 	if err != nil {
-		return nil, fmt.Errorf("error instantiating non-worker split listener: %w", err)
+		return nil, fmt.Errorf("error instantiating reverse grpc split listener: %w", err)
 	}
 
 	// This wraps the reverse grpc pki worker connections with a listener which
@@ -180,6 +185,20 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 		return nil, fmt.Errorf("%s: error creating reverse grpc pki worker tracking listener: %w", op, err)
 	}
 
+	// Connections coming in here are authed by nodeenrollment and are for the
+	// multi-hop session-proxying
+	dataPlaneProxyListener, err := w.workerAuthSplitListener.GetListener(common.DataPlaneProxyAlpnValue, nodee.WithNativeConns(true))
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating websocket proxying split listener: %w", err)
+	}
+
+	// This wraps the web socket proxying worker connections with a listener which
+	// adds the worker key id of the connections to the worker's pkiConnManager.
+	dataPlaneProxyTrackingListener, err := cluster.NewTrackingListener(w.baseContext, dataPlaneProxyListener, w.pkiConnManager)
+	if err != nil {
+		return nil, fmt.Errorf("%s: error creating web socket proxying tracking listener: %w", op, err)
+	}
+
 	statsHandler, err := metric.InstrumentClusterStatsHandler(w.baseContext)
 	if err != nil {
 		return nil, errors.Wrap(w.baseContext, err, op)
@@ -214,10 +233,11 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 	}
 
 	return func() {
+		handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, "reverse-grpc"),
+			metric.InstrumentWorkerClusterTrackingListener(dataPlaneProxyTrackingListener, "multihop-proxy-dataplane"), w.downstreamRoutes)
 		go w.workerAuthSplitListener.Start()
 		go httpServer.Serve(proxyListener)
 		go ln.GrpcServer.Serve(metric.InstrumentWorkerClusterTrackingListener(pkiWorkerTrackingListener, "grpc"))
-		go handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, "reverse-grpc"), w.downstreamRoutes, -1)
 	}, nil
 }
 

internal/daemon/worker/status.go

@@ -21,7 +21,7 @@ import (
 
 var firstStatusCheckPostHooks []func(context.Context, *Worker) error
 
-var downstreamWorkersFactory func(ctx context.Context, workerId string) (downstreamers, error)
+var downstreamWorkersFactory func(ctx context.Context, workerId string, ver string) (downstreamers, error)
 
 type LastStatusInformation struct {
 	*pbs.StatusResponse
@@ -305,7 +305,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 	// If we have post hooks for after the first status check, run them now
 	if w.everAuthenticated.CAS(authenticationStatusFirstAuthentication, authenticationStatusFirstStatusRpcSuccessful) {
 		if downstreamWorkersFactory != nil {
-			w.downstreamWorkers, err = downstreamWorkersFactory(cancelCtx, w.LastStatusSuccess().WorkerId)
+			w.downstreamWorkers, err = downstreamWorkersFactory(cancelCtx, w.LastStatusSuccess().WorkerId, versionInfo.FullVersionNumber(false))
 			if err != nil {
 				event.WriteError(cancelCtx, op, err)
 				w.conf.ServerSideShutdownCh <- struct{}{}
@@ -349,10 +349,10 @@ func (w *Worker) updateAddresses(cancelCtx context.Context, addrs []string, addr
 	}
 
 	// regardless of whether or not it's a new address, we need to set
-	// them for dialingListeners
+	// them for secondary connections
 	for _, as := range *addressReceivers {
 		switch {
-		case as.Type() == dialingListenerReceiverType:
+		case as.Type() == secondaryConnectionReceiverType:
 			tmpAddrs := make([]string, len(addrs))
 			copy(tmpAddrs, addrs)
 			if len(tmpAddrs) == 0 {

internal/daemon/worker/worker.go

@@ -53,7 +53,7 @@ type downstreamRouter interface {
 	// ProcessPendingConnections starts a function that continually processes
 	// incoming client connections. This only returns when the provided context
 	// is done.
-	StartProcessingPendingConnections(context.Context, func() string)
+	StartProcessingPendingConnections(context.Context, func() string) error
 }
 
 // downstreamers provides at least a minimum interface that must be met by a
@@ -446,7 +446,9 @@ func (w *Worker) Start() error {
 		}
 		go func() {
 			defer w.tickerWg.Done()
-			w.downstreamRoutes.StartProcessingPendingConnections(w.baseContext, servNameFn)
+			if err := w.downstreamRoutes.StartProcessingPendingConnections(w.baseContext, servNameFn); err != nil {
+				errors.Wrap(w.baseContext, err, op)
+			}
 		}()
 		go func() {
 			defer w.tickerWg.Done()