fix(server): raise topic publish queue capacity

deploy/argocd/im-server-test/statefulset-performance-patch.yaml

@@ -32,6 +32,12 @@ spec:
               value: "3"
             - name: CLUSTER_ROUTE_RETRY_BACKOFF_MS
               value: "25"
+            - name: TOPIC_CLIENT_QUEUE_BUFFER
+              value: "4096"
+            - name: TOPIC_SERVER_QUEUE_BUFFER
+              value: "512"
+            - name: TOPIC_PUBLISH_ENQUEUE_WAIT_MS
+              value: "200"
             - name: MYSQL_MAX_OPEN_CONNS
               value: "24"
             - name: MYSQL_MAX_IDLE_CONNS

docker/README.md

@@ -193,6 +193,9 @@ You can specify the following environment variables when issuing `docker run` co
 | `CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS` | number | `300` | Maximum wait for a full proxy-to-master shard before returning 503. |
 | `CLUSTER_ROUTE_RETRIES` | number | `3` | Number of retries after a remote server-message route is rejected as overloaded. |
 | `CLUSTER_ROUTE_RETRY_BACKOFF_MS` | number | `25` | Initial exponential retry delay for remote server-message routes. |
+| `TOPIC_CLIENT_QUEUE_BUFFER` | number | `4096` | Per-topic client publish queue capacity. Increase for large groups with bursty publishers. |
+| `TOPIC_SERVER_QUEUE_BUFFER` | number | `512` | Per-topic server-message queue capacity. |
+| `TOPIC_PUBLISH_ENQUEUE_WAIT_MS` | number | `200` | Maximum wait for an attached session publish to enter a full topic queue before returning 503. |
 | `DEBUG_EMAIL_VERIFICATION_CODE`[^2] | string |  | Enable dummy email verification code, e.g. `123456`. Disabled by default (empty string). |
 | `DEFAULT_COUNTRY_CODE`[^2] | string | `US` | 2-letter country code to assign to sessions by default when the country isn't specified by the client explicitly and it's impossible to infer it. |
 | `EXT_CONFIG`[^1] | string |  | Path to external config file to use instead of the built-in one. If this parameter is used, most other variables are ignored[^1]. |

server/hub.go

@@ -10,6 +10,8 @@
 package main
 
 import (
+	"os"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -30,6 +32,44 @@ var requestLatencyDistribution = []float64{1, 2, 3, 4, 5, 6, 8, 10, 13, 16, 20,
 var outgoingMessageSizeDistribution = []float64{1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 16384,
 	65536, 262144, 1048576, 4194304, 16777216, 67108864, 268435456, 1073741824, 4294967296}
 
+const (
+	defaultTopicClientQueue    = 4096
+	defaultTopicServerQueue    = 512
+	defaultTopicPublishWait    = 200 * time.Millisecond
+	topicClientQueueEnv        = "TOPIC_CLIENT_QUEUE_BUFFER"
+	topicServerQueueEnv        = "TOPIC_SERVER_QUEUE_BUFFER"
+	topicPublishEnqueueWaitEnv = "TOPIC_PUBLISH_ENQUEUE_WAIT_MS"
+)
+
+var topicQueueConfig = loadTopicQueueConfig()
+
+type topicQueueRuntimeConfig struct {
+	clientBuffer       int
+	serverBuffer       int
+	publishEnqueueWait time.Duration
+}
+
+func loadTopicQueueConfig() topicQueueRuntimeConfig {
+	return topicQueueRuntimeConfig{
+		clientBuffer:       intEnvOrDefault(topicClientQueueEnv, defaultTopicClientQueue),
+		serverBuffer:       intEnvOrDefault(topicServerQueueEnv, defaultTopicServerQueue),
+		publishEnqueueWait: time.Duration(intEnvOrDefault(topicPublishEnqueueWaitEnv, int(defaultTopicPublishWait/time.Millisecond))) * time.Millisecond,
+	}
+}
+
+func intEnvOrDefault(name string, fallback int) int {
+	raw := strings.TrimSpace(os.Getenv(name))
+	if raw == "" {
+		return fallback
+	}
+	parsed, err := strconv.Atoi(raw)
+	if err != nil || parsed <= 0 {
+		logs.Warn.Printf("invalid %s=%q, using %d", name, raw, fallback)
+		return fallback
+	}
+	return parsed
+}
+
 // Request to hub to remove the topic
 type topicUnreg struct {
 	// Tenant which owns the topic.
@@ -261,8 +301,8 @@ func (h *Hub) run() {
 					// Indicates a proxy topic.
 					isProxy:   globals.cluster.isRemoteTopic(key),
 					sessions:  make(map[*Session]perSessionData),
-					clientMsg: make(chan *ClientComMessage, 192),
-					serverMsg: make(chan *ServerComMessage, 64),
+					clientMsg: make(chan *ClientComMessage, topicQueueConfig.clientBuffer),
+					serverMsg: make(chan *ServerComMessage, topicQueueConfig.serverBuffer),
 					reg:       make(chan *ClientComMessage, 256),
 					unreg:     make(chan *ClientComMessage, 256),
 					meta:      make(chan *ClientComMessage, 64),

server/session.go

@@ -1019,6 +1019,24 @@ func (s *Session) publish(msg *ClientComMessage) {
 	if sub := s.getSub(msg.RcptTo); sub != nil {
 		// This is a post to a subscribed topic. The message is sent to the topic only
 		msg.queuedAt = time.Now()
+		if topicQueueConfig.publishEnqueueWait > 0 {
+			timer := time.NewTimer(topicQueueConfig.publishEnqueueWait)
+			select {
+			case sub.broadcast <- msg:
+				immetrics.ObserveInternalQueueEnqueue("session_topic_publish", len(sub.broadcast), cap(sub.broadcast), true)
+				if !timer.Stop() {
+					select {
+					case <-timer.C:
+					default:
+					}
+				}
+			case <-timer.C:
+				immetrics.ObserveInternalQueueEnqueue("session_topic_publish", len(sub.broadcast), cap(sub.broadcast), false)
+				// Reply with a 503 to the user.
+				s.queueOut(ErrServiceUnavailableReply(msg, msg.Timestamp))
+				logs.Err.Println("s.publish: sub.broadcast channel full, topic ", msg.RcptTo, s.sid)
+			}
+		} else {
 			select {
 			case sub.broadcast <- msg:
 				immetrics.ObserveInternalQueueEnqueue("session_topic_publish", len(sub.broadcast), cap(sub.broadcast), true)
@@ -1028,6 +1046,7 @@ func (s *Session) publish(msg *ClientComMessage) {
 				s.queueOut(ErrServiceUnavailableReply(msg, msg.Timestamp))
 				logs.Err.Println("s.publish: sub.broadcast channel full, topic ", msg.RcptTo, s.sid)
 			}
+		}
 	} else if msg.RcptTo == "sys" {
 		if s.authLvl != auth.LevelRoot {
 			s.queueOut(ErrPermissionDeniedReply(msg, msg.Timestamp))