Don't allow session sync to block channel. Fixes #1849

openziti · Mar 20, 2024 · 1e17a70 · 1e17a70
1 parent 37e5394
commit 1e17a70
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 3 deletions.
diff --git a/controller/server/controller.go b/controller/server/controller.go
@@ -234,7 +234,7 @@ func (c *Controller) Initialize() {
 	//after InitPersistence
 	c.AppEnv.Broker = env.NewBroker(c.AppEnv, sync2.NewInstantStrategy(c.AppEnv, sync2.InstantStrategyOptions{
 		MaxQueuedRouterConnects:  100,
-		MaxQueuedClientHellos:    100,
+		MaxQueuedClientHellos:    1000,
 		RouterConnectWorkerCount: 10,
 		SyncWorkerCount:          10,
 		RouterTxBufferSize:       100,

diff --git a/controller/sync_strats/sync_instant.go b/controller/sync_strats/sync_instant.go
@@ -528,7 +528,34 @@ func (strategy *InstantStrategy) ReceiveResync(routerId string, _ *edge_ctrl_pb.
 
 	rtx.RouterModelIndex = nil
 
-	strategy.receivedClientHelloQueue <- rtx
+	strategy.queueClientHello(rtx)
+}
+
+func (strategy *InstantStrategy) queueClientHello(rtx *RouterSender) {
+	select {
+	case strategy.receivedClientHelloQueue <- rtx:
+		return
+	default:
+	}
+
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	go func() {
+		for {
+			if ch := rtx.Router.Control; ch == nil || ch.IsClosed() {
+				return
+			}
+
+			select {
+			case strategy.receivedClientHelloQueue <- rtx:
+				return
+			case <-strategy.stopNotify:
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
 }
 
 func (strategy *InstantStrategy) ReceiveClientHello(routerId string, msg *channel.Message, respHello *edge_ctrl_pb.ClientHello) {
@@ -589,7 +616,7 @@ func (strategy *InstantStrategy) ReceiveClientHello(routerId string, msg *channe
 
 	serverVersion := build.GetBuildInfo().Version()
 	logger.Infof("edge router sent hello with version [%s] to controller with version [%s]", respHello.Version, serverVersion)
-	strategy.receivedClientHelloQueue <- rtx
+	strategy.queueClientHello(rtx)
 }
 
 func (strategy *InstantStrategy) synchronize(rtx *RouterSender) {