Add interfaces and error codes for processing downstream router connections (#2511)

* Add interfaces and error codes for processing downstream router connections

internal/daemon/controller/controller.go

@@ -49,6 +49,11 @@ type downstreamRouter interface {
 	// StartRouteMgmtTicking starts a ticker which manages the router's
 	// connections.
 	StartRouteMgmtTicking(context.Context, func() string, int) error
+
+	// ProcessPendingConnections starts a function that continually processes
+	// incoming client connections. This only returns when the provided context
+	// is done.
+	StartProcessingPendingConnections(context.Context, func() string)
 }
 
 // downstreamWorkersTicker defines an interface for a ticker that maintains the
@@ -425,19 +430,25 @@ func (c *Controller) Start() error {
 	}()
 
 	if c.downstreamRoutes != nil {
-		c.tickerWg.Add(1)
+		c.tickerWg.Add(2)
+
+		servNameFn := func() string {
+			switch {
+			case c.conf.RawConfig.Controller.Name != "":
+				return c.conf.RawConfig.Controller.Name
+			default:
+				return "unknown controller name"
+			}
+		}
+		go func() {
+			defer c.tickerWg.Done()
+			c.downstreamRoutes.StartProcessingPendingConnections(c.baseContext, servNameFn)
+		}()
 		go func() {
 			defer c.tickerWg.Done()
 			err := c.downstreamRoutes.StartRouteMgmtTicking(
 				c.baseContext,
-				func() string {
-					switch {
-					case c.conf.RawConfig.Controller.Name != "":
-						return c.conf.RawConfig.Controller.Name
-					default:
-						return "unknown controller name"
-					}
-				},
+				servNameFn,
 				-1,
 			)
 			if err != nil {

internal/daemon/worker/worker.go

@@ -46,6 +46,11 @@ type downstreamRouter interface {
 	// StartRouteMgmtTicking starts a ticker which manages the router's
 	// connections.
 	StartRouteMgmtTicking(context.Context, func() string, int) error
+
+	// ProcessPendingConnections starts a function that continually processes
+	// incoming client connections. This only returns when the provided context
+	// is done.
+	StartProcessingPendingConnections(context.Context, func() string)
 }
 
 // downstreamers provides at least a minimum interface that must be met by a
@@ -389,7 +394,7 @@ func (w *Worker) Start() error {
 	// Rather than deal with some of the potential error conditions for Add on
 	// the waitgroup vs. Done (in case a function exits immediately), we will
 	// always start rotation and simply exit early if we're using KMS
-	w.tickerWg.Add(3)
+	w.tickerWg.Add(2)
 	go func() {
 		defer w.tickerWg.Done()
 		w.startStatusTicking(w.baseContext, w.sessionManager, &w.addressReceivers)
@@ -398,24 +403,31 @@ func (w *Worker) Start() error {
 		defer w.tickerWg.Done()
 		w.startAuthRotationTicking(w.baseContext)
 	}()
-	go func() {
-		defer w.tickerWg.Done()
-		if w.downstreamRoutes != nil {
+
+	if w.downstreamRoutes != nil {
+		w.tickerWg.Add(2)
+		servNameFn := func() string {
+			if s := w.LastStatusSuccess(); s != nil {
+				return s.WorkerId
+			}
+			return "unknown worker id"
+		}
+		go func() {
+			defer w.tickerWg.Done()
+			w.downstreamRoutes.StartProcessingPendingConnections(w.baseContext, servNameFn)
+		}()
+		go func() {
+			defer w.tickerWg.Done()
 			err := w.downstreamRoutes.StartRouteMgmtTicking(
 				w.baseContext,
-				func() string {
-					if s := w.LastStatusSuccess(); s != nil {
-						return s.WorkerId
-					}
-					return "unknown worker id"
-				},
+				servNameFn,
 				-1, // indicates the ticker should run until cancelled.
 			)
 			if err != nil {
 				errors.Wrap(w.baseContext, err, op)
 			}
-		}
-	}()
+		}()
+	}
 
 	w.workerStartTime = time.Now()
 	w.started.Store(true)

internal/errors/code.go

@@ -45,6 +45,10 @@ const (
 	CycleFound                         = 121 // CycleFound represents an error when a cycle is found between a parent and child worker
 	WorkerConnNotFound                 = 122 // WorkerConnNotFound represents an error when a connection to a worker is not found
 	KmsWorkerUnsupportedOperation      = 123 // KmsWorkerUnsupportedOperation represents an error when a KMS worker is not supported for an operation
+	// Note: Currently unused in OSS
+	RetryLimitExceeded = 124 // RetryLimitExceeded represents an error when a retry limit is exceeded
+	// Note: Currently unused in OSS
+	QueueIsFull = 125 // QueueIsFull results in attempting to add an item to a queue which is full
 
 	AuthAttemptExpired Code = 198 // AuthAttemptExpired represents an expired authentication attempt
 	AuthMethodInactive Code = 199 // AuthMethodInactive represents an error that means the auth method is not active.

internal/errors/code_test.go

@@ -337,6 +337,16 @@ func TestCode_Both_String_Info(t *testing.T) {
 			c:    KmsWorkerUnsupportedOperation,
 			want: KmsWorkerUnsupportedOperation,
 		},
+		{
+			name: "RetryLimitExceeded",
+			c:    RetryLimitExceeded,
+			want: RetryLimitExceeded,
+		},
+		{
+			name: "QueueIsFull",
+			c:    QueueIsFull,
+			want: QueueIsFull,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

internal/errors/info.go

@@ -272,4 +272,12 @@ var errorCodeInfo = map[Code]Info{
 		Message: "unsupported operation for a kms worker",
 		Kind:    State,
 	},
+	QueueIsFull: {
+		Message: "queue is full",
+		Kind:    State,
+	},
+	RetryLimitExceeded: {
+		Message: "retry limit exceeded",
+		Kind:    State,
+	},
 }