Lift link fault handling off the control channel receive goroutine

A fault was handled by spawning a goroutine per message, so a burst of faults
spawned without limit. Handling them inline instead would put the burst on the
receive goroutine, which carries every other default priority message on that
underlay.

- adds a pool for work lifted off a control channel's receive goroutine, kept separate from the rate limiter pool, whose workers park waiting for controller send capacity; sharing it would make receiving depend on the very back-pressure that lifting work off the receive goroutine exists to survive
- hands fault handling to that pool, and falls back to spawning when the pool is full rather than dropping, since a dropped fault leaves a dead link marked up with no repair loop behind it
- counts the fallbacks, so the pool being too small is visible rather than silent
- sizes the queue generously and the worker count modestly: the work arrives in bursts, one per controller reconnect, but is per control channel and there are only ever a handful of those

router/env/config_forwarder.go

@@ -23,6 +23,13 @@ import (
 )
 
 const (
+	// DefaultRxPoolQueueLength and DefaultRxPoolWorkerCount size the pool for work lifted off a control
+	// channel's receive goroutine. The queue is generous because the work arrives in bursts, one per
+	// controller reconnect, while the worker count stays modest since the work is per control channel and
+	// there are only ever a handful of those.
+	DefaultRxPoolQueueLength = 1024
+	DefaultRxPoolWorkerCount = 16
+
 	DefaultXgressCloseCheckInterval    = 5 * time.Second
 	DefaultXgressDialDwellTime         = 0
 	DefaultFaultTxInterval             = 15 * time.Second

router/env/env.go

@@ -49,6 +49,11 @@ type RouterEnv interface {
 	RenderJsonConfig() (string, error)
 	GetHeartbeatOptions() HeartbeatOptions
 	GetRateLimiterPool() goroutines.Pool
+	// GetRxPool returns the pool for work lifted off a control channel's receive goroutine. Distinct from
+	// the rate limiter pool, whose workers park waiting for controller send capacity: sharing that pool
+	// would make receiving depend on the very back-pressure that lifting work off the receive goroutine is
+	// meant to survive.
+	GetRxPool() goroutines.Pool
 	GetCtrlRateLimiter() rate.AdaptiveRateLimitTracker
 	GetVersionInfo() versions.VersionProvider
 	GetRouterDataModel() *common.RouterDataModel

router/handler_ctrl/bind.go

@@ -116,7 +116,7 @@ func (self *bindHandler) BindChannel(binding channel.Binding) error {
 	channel.AddReceiveHandlers(binding, self.env.GetInspectHandler())
 	channel.AddReceiveHandlers(binding, newCheckStaleLinksHandler(self.env))
 	channel.AddReceiveHandlers(binding, newSettingsHandler(self.env))
-	channel.AddReceiveHandlers(binding, newFaultHandler(self.env.GetXlinkRegistry()))
+	channel.AddReceiveHandlers(binding, newFaultHandler(self.env))
 	channel.AddReceiveHandlers(binding, self.ctrlAddrChangeHandler)
 	channel.AddReceiveHandlers(binding, self.clusterLeaderChangeHandler)
 

router/handler_ctrl/fault.go

@@ -19,17 +19,25 @@ package handler_ctrl
 import (
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/channel/v5"
+	"github.com/openziti/metrics"
 	"github.com/openziti/ziti/v2/common/pb/ctrl_pb"
+	"github.com/openziti/ziti/v2/router/env"
 	"github.com/openziti/ziti/v2/router/xlink"
 	"google.golang.org/protobuf/proto"
 )
 
 type faultHandler struct {
 	xlinkRegistry xlink.Registry
+	env           env.RouterEnv
+	poolFallbacks metrics.Meter
 }
 
-func newFaultHandler(registry xlink.Registry) *faultHandler {
-	return &faultHandler{xlinkRegistry: registry}
+func newFaultHandler(routerEnv env.RouterEnv) *faultHandler {
+	return &faultHandler{
+		xlinkRegistry: routerEnv.GetXlinkRegistry(),
+		env:           routerEnv,
+		poolFallbacks: routerEnv.GetMetricsRegistry().Meter("link.fault.rx_pool_fallback"),
+	}
 }
 
 func (self *faultHandler) ContentType() int32 {
@@ -45,7 +53,19 @@ func (self *faultHandler) HandleReceive(msg *channel.Message, ch channel.Channel
 		return
 	}
 
-	go self.handleFault(msg, ch, fault)
+	handle := func() { self.handleFault(msg, ch, fault) }
+
+	// Bounded while the receive pool has room, unbounded when it does not. Spawning is the fallback rather
+	// than the default because a dropped fault leaves a dead link marked up with no repair loop behind it,
+	// and because waiting for pool capacity would put the burst on this goroutine, which also carries every
+	// other default priority message on this underlay. The pool is resolved here rather than at construction
+	// since handlers are built before the pools exist.
+	if pool := self.env.GetRxPool(); pool != nil && pool.QueueOrError(handle) == nil {
+		return
+	}
+
+	self.poolFallbacks.Mark(1)
+	go handle()
 }
 
 func (self *faultHandler) handleFault(_ *channel.Message, ch channel.Channel, fault *ctrl_pb.Fault) {

router/router.go

@@ -92,6 +92,7 @@ type Router struct {
 	config              *env.Config
 	routerCapabilities  *capabilities.RouterCapabilityMask
 	ctrls               env.NetworkControllers
+	rxPool              goroutines.Pool
 	ctrlBindhandler     channel.BindHandler
 	faulter             *forwarder.Faulter
 	forwarder           *forwarder.Forwarder
@@ -667,6 +668,10 @@ func (self *Router) initGoroutinePools() error {
 		return err
 	}
 
+	if err := self.initRxPool(); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -730,6 +735,37 @@ func (self *Router) GetLinkDialerPool() goroutines.Pool {
 	return self.linkDialerPool
 }
 
+// initRxPool builds the pool for work lifted off a control channel's receive goroutine. It is deliberately
+// separate from the rate limiter pool: that pool's workers park waiting for controller send capacity, so
+// sharing it would leave receiving to compete with, and be starved by, the back-pressure that moving work
+// off the receive goroutine exists to survive.
+func (self *Router) initRxPool() error {
+	rxPoolConfig := goroutines.PoolConfig{
+		QueueSize:   env.DefaultRxPoolQueueLength,
+		MinWorkers:  0,
+		MaxWorkers:  env.DefaultRxPoolWorkerCount,
+		IdleTime:    30 * time.Second,
+		CloseNotify: self.GetCloseNotify(),
+		PanicHandler: func(err interface{}) {
+			pfxlog.Logger().WithField(logrus.ErrorKey, err).WithField("backtrace", string(debug.Stack())).Error("panic during control channel receive processing")
+		},
+	}
+
+	servermetrics.ConfigureGoroutinesPoolMetrics(&rxPoolConfig, self.GetMetricsRegistry(), "pool.router.rx")
+
+	rxPool, err := goroutines.NewPool(rxPoolConfig)
+	if err != nil {
+		return errors.Wrap(err, "error creating control channel receive pool")
+	}
+
+	self.rxPool = rxPool
+	return nil
+}
+
+func (self *Router) GetRxPool() goroutines.Pool {
+	return self.rxPool
+}
+
 func (self *Router) GetRateLimiterPool() goroutines.Pool {
 	return self.rateLimiterPool
 }