perf(group): reduce websocket fanout serialization

server/cluster.go

@@ -284,6 +284,9 @@ type ClusterResp struct {
 
 	// Original request type.
 	OrigReqType ProxyReqType
+
+	// Set after RPC decoding on the receiving node; not serialized by gob.
+	proxyQueuedAt time.Time
 }
 
 // ClusterPing is used to detect node restarts.
@@ -743,6 +746,7 @@ func (Cluster) TopicProxy(msg *ClusterResp, unused *bool) error {
 	// Find appropriate topic, send the message to it.
 	if t := globals.hub.topicGet(types.TopicKey{TenantID: msg.TenantID, Topic: msg.RcptTo}); t != nil {
 		msg.SrvMsg.uid = types.ParseUserId(msg.SrvMsg.AsUser)
+		msg.proxyQueuedAt = time.Now()
 		select {
 		case t.proxy <- msg:
 			immetrics.ObserveInternalQueueEnqueue("cluster_topic_proxy", len(t.proxy), cap(t.proxy), true)
@@ -1420,6 +1424,8 @@ func (sess *Session) clusterWriteLoop(forTopic string) {
 				return
 			}
 			srvMsg := msg.(*ServerComMessage)
+			immetrics.ObserveClusterM2PDequeue(sess.clnode.name, srvMsg.clusterQueuedAt)
+			srvMsg.clusterQueuedAt = time.Time{}
 			response := &ClusterResp{TenantID: srvMsg.TenantID, SrvMsg: srvMsg}
 			if srvMsg.sess == nil {
 				response.OrigSid = "*"

server/datamodel.go

@@ -881,6 +881,8 @@ type ServerComMessage struct {
 	PreserveTopicName bool `json:"-"`
 	// User id affected by this message.
 	uid types.Uid
+	// Timestamp when this message entered a master-to-proxy multiplex queue.
+	clusterQueuedAt time.Time
 }
 
 // Deep-shallow copy of ServerComMessage. Deep copy of service fields,
@@ -899,6 +901,7 @@ func (src *ServerComMessage) copy() *ServerComMessage {
 		SkipSid:           src.SkipSid,
 		PreserveTopicName: src.PreserveTopicName,
 		uid:               src.uid,
+		clusterQueuedAt:   src.clusterQueuedAt,
 	}
 
 	dst.Ctrl = src.Ctrl.copy()

server/hdl_websock.go

@@ -105,7 +105,9 @@ func (sess *Session) readLoop() {
 		atomic.StoreInt64(&sess.lastAction, types.TimeNow().UnixNano())
 		_ = sess.ws.SetReadDeadline(time.Now().Add(pongWait))
 		statsInc("IncomingMessagesWebsockTotal", 1)
+		dispatchStarted := time.Now()
 		sess.dispatchRaw(raw)
+		immetrics.ObserveWSReadDispatch(dispatchStarted)
 	}
 }
 
@@ -160,6 +162,11 @@ func (sess *Session) writeLoop() {
 				if isAcceptedCtrl(v) {
 					immetrics.ObserveCtrlACKDelivery("write", v.Timestamp)
 				}
+			case sharedServerMessage:
+				statsAddHistSample("OutgoingMessageSize", float64(len(v)))
+				if !sess.sendMessage([]byte(v)) {
+					return
+				}
 			default: // serialized message
 				if !sess.sendMessage(v) {
 					return

server/metrics/metrics.go

@@ -190,6 +190,45 @@ var (
 		Help:    "Time a client publish waited in a topic queue before processing by topic kind.",
 		Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
 	}, []string{"pod", "kind"})
+	TopicProcessDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_topic_process_duration_seconds",
+		Help:    "End-to-end topic event processing duration by pod, topic kind, role, and operation.",
+		Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
+	}, []string{"pod", "kind", "role", "operation"})
+	TopicBroadcastDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_topic_broadcast_duration_seconds",
+		Help:    "Time spent preparing and enqueueing one topic broadcast by pod, topic kind, and role.",
+		Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
+	}, []string{"pod", "kind", "role"})
+	TopicBroadcastTargets = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_topic_broadcast_targets",
+		Help:    "Number of local WebSocket, multiplex, or other targets considered by one topic broadcast.",
+		Buckets: []float64{0, 1, 10, 50, 100, 250, 500, 750, 1000, 1500, 2000, 3000},
+	}, []string{"pod", "kind", "role", "target_type"})
+	TopicSharedSerializationTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "im_topic_shared_serialization_total",
+		Help: "Shared WebSocket serialization events by pod, topic kind, and result.",
+	}, []string{"pod", "kind", "result"})
+	ClusterM2PQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_cluster_m2p_queue_wait_duration_seconds",
+		Help:    "Time a master-to-proxy response waited in the multiplex outbound queue by pod and peer.",
+		Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
+	}, []string{"pod", "peer"})
+	ClusterM2PQueueDepth = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_cluster_m2p_queue_depth",
+		Help:    "Master-to-proxy multiplex outbound queue depth observed after enqueue by pod and peer.",
+		Buckets: []float64{0, 1, 4, 8, 16, 32, 64, 96, 128, 160},
+	}, []string{"pod", "peer"})
+	ClusterProxyQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_cluster_proxy_queue_wait_duration_seconds",
+		Help:    "Time a cluster response waited in the proxy topic queue by pod and message type.",
+		Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
+	}, []string{"pod", "message_type"})
+	WSReadDispatchDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "im_ws_read_dispatch_duration_seconds",
+		Help:    "Time from a completed WebSocket frame read through JSON decoding and local dispatch.",
+		Buckets: []float64{0.0001, 0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5},
+	}, []string{"pod"})
 	PushDispatchQueueEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: "im_push_dispatch_queue_enqueue_total",
 		Help: "Asynchronous push dispatch queue enqueue attempts by pod and result.",
@@ -242,6 +281,8 @@ func init() {
 		ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration,
 		ClusterRouteTotal, ClusterProxyResponseTotal, InternalQueueEnqueueTotal, InternalQueueDepth,
 		InternalQueueCapacity, InternalQueueCurrent, MessagePersistDuration, MessagePublishStageDuration, TopicQueueWaitDuration,
+		TopicProcessDuration, TopicBroadcastDuration, TopicBroadcastTargets, TopicSharedSerializationTotal,
+		ClusterM2PQueueWaitDuration, ClusterM2PQueueDepth, ClusterProxyQueueWaitDuration, WSReadDispatchDuration,
 		PushDispatchQueueEnqueueTotal, PushDispatchQueueDepth, PushDispatchQueueWaitDuration,
 		PushDispatchDuration, PushRecipientCount,
 		KafkaMessagePersistTotal)
@@ -279,6 +320,47 @@ func ObserveTopicQueueWait(kind string, queuedAt time.Time) {
 	TopicQueueWaitDuration.WithLabelValues(metricPodName(), normalizeLabel(kind)).Observe(time.Since(queuedAt).Seconds())
 }
 
+func ObserveTopicProcess(kind, role, operation string, started time.Time) {
+	TopicProcessDuration.WithLabelValues(metricPodName(), normalizeLabel(kind), normalizeLabel(role),
+		normalizeLabel(operation)).Observe(time.Since(started).Seconds())
+}
+
+func ObserveTopicBroadcast(kind, role string, started time.Time, websocketTargets, multiplexTargets, otherTargets int) {
+	pod := metricPodName()
+	kind = normalizeLabel(kind)
+	role = normalizeLabel(role)
+	TopicBroadcastDuration.WithLabelValues(pod, kind, role).Observe(time.Since(started).Seconds())
+	TopicBroadcastTargets.WithLabelValues(pod, kind, role, "websocket").Observe(float64(websocketTargets))
+	TopicBroadcastTargets.WithLabelValues(pod, kind, role, "multiplex").Observe(float64(multiplexTargets))
+	TopicBroadcastTargets.WithLabelValues(pod, kind, role, "other").Observe(float64(otherTargets))
+}
+
+func IncTopicSharedSerialization(kind, result string) {
+	TopicSharedSerializationTotal.WithLabelValues(metricPodName(), normalizeLabel(kind), normalizeLabel(result)).Inc()
+}
+
+func ObserveClusterM2PEnqueue(peer string, depth int) {
+	ClusterM2PQueueDepth.WithLabelValues(metricPodName(), normalizeLabel(peer)).Observe(float64(depth))
+}
+
+func ObserveClusterM2PDequeue(peer string, queuedAt time.Time) {
+	if queuedAt.IsZero() {
+		return
+	}
+	ClusterM2PQueueWaitDuration.WithLabelValues(metricPodName(), normalizeLabel(peer)).Observe(time.Since(queuedAt).Seconds())
+}
+
+func ObserveClusterProxyQueueWait(messageType string, queuedAt time.Time) {
+	if queuedAt.IsZero() {
+		return
+	}
+	ClusterProxyQueueWaitDuration.WithLabelValues(metricPodName(), normalizeLabel(messageType)).Observe(time.Since(queuedAt).Seconds())
+}
+
+func ObserveWSReadDispatch(started time.Time) {
+	WSReadDispatchDuration.WithLabelValues(metricPodName()).Observe(time.Since(started).Seconds())
+}
+
 func ObservePushDispatchEnqueue(depth int, accepted bool) {
 	result := "ok"
 	if !accepted {

server/monitoring_http_test.go

@@ -98,6 +98,13 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) {
 	immetrics.ObserveWSSendQueueDequeue(time.Now(), false)
 	immetrics.ObserveCtrlACKDelivery("write", time.Now())
 	immetrics.ObserveTopicQueueWait("group", time.Now())
+	immetrics.ObserveTopicProcess("group", "master", "publish", time.Now())
+	immetrics.ObserveTopicBroadcast("group", "master", time.Now(), 1, 0, 0)
+	immetrics.IncTopicSharedSerialization("group", "serialized")
+	immetrics.ObserveClusterM2PEnqueue("peer", 1)
+	immetrics.ObserveClusterM2PDequeue("peer", time.Now())
+	immetrics.ObserveClusterProxyQueueWait("data", time.Now())
+	immetrics.ObserveWSReadDispatch(time.Now())
 	immetrics.ObservePushDispatchEnqueue(1, true)
 	immetrics.ObservePushDispatch(time.Now(), 1, time.Now(), nil)
 	wrt := httptest.NewRecorder()
@@ -117,7 +124,11 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) {
 		"im_ctrl_ack_delivery_duration_seconds",
 		"im_cluster_p2m_enqueue_total", "im_cluster_rpc_duration_seconds", "im_cluster_route_total",
 		"im_internal_queue_enqueue_total", "im_internal_queue_depth", "im_internal_queue_capacity", "im_internal_queue_current",
-		"im_topic_queue_wait_duration_seconds", "im_push_dispatch_queue_enqueue_total",
+		"im_topic_queue_wait_duration_seconds", "im_topic_process_duration_seconds",
+		"im_topic_broadcast_duration_seconds", "im_topic_broadcast_targets", "im_topic_shared_serialization_total",
+		"im_cluster_m2p_queue_wait_duration_seconds", "im_cluster_m2p_queue_depth",
+		"im_cluster_proxy_queue_wait_duration_seconds", "im_ws_read_dispatch_duration_seconds",
+		"im_push_dispatch_queue_enqueue_total",
 		"im_push_dispatch_queue_depth", "im_push_dispatch_queue_wait_duration_seconds",
 		"im_push_dispatch_duration_seconds", "im_push_recipient_count",
 		"im_kafka_message_persist_total", "im_kafka_consumer_lag", "im_kafka_dead_letters_total"} {

server/session.go

@@ -195,6 +195,9 @@ type Subscription struct {
 	supd chan<- *sessionUpdate
 }
 
+// sharedServerMessage is an immutable JSON payload reused by multiple WebSocket sessions.
+type sharedServerMessage []byte
+
 func (s *Session) enqueueOutbound(payload any) bool {
 	if s.isMultiplex() {
 		return s.enqueueMultiplexOutbound(payload, multiplexEnqueueWait)
@@ -234,8 +237,16 @@ func (s *Session) enqueueMultiplexOutbound(payload any, wait time.Duration) bool
 			s.clusterQueueMu.Unlock()
 			return true
 		}
+		if msg, ok := payload.(*ServerComMessage); ok {
+			msg.clusterQueuedAt = time.Now()
+		}
 		select {
 		case s.send <- payload:
+			peer := "unknown"
+			if s.clnode != nil {
+				peer = s.clnode.name
+			}
+			immetrics.ObserveClusterM2PEnqueue(peer, len(s.send))
 			s.clusterQueueMu.Unlock()
 			return true
 		default:
@@ -525,6 +536,17 @@ func (s *Session) queueOutBytes(data []byte) bool {
 	return true
 }
 
+func (s *Session) queueOutSharedBytes(data []byte) bool {
+	if s == nil || atomic.LoadInt32(&s.terminating) > 0 {
+		return true
+	}
+	if !s.enqueueOutbound(sharedServerMessage(data)) {
+		logs.Err.Println("s.queueOutSharedBytes: session's send queue full", s.sid)
+		return false
+	}
+	return true
+}
+
 func (s *Session) maybeScheduleClusterWriteLoop() {
 	if s.multi != nil {
 		s.multi.scheduleClusterWriteLoop()

server/topic.go

@@ -650,12 +650,20 @@ func (t *Topic) runLocal(hub *Hub) {
 
 // handleClientMsg is the top-level handler of messages received by the topic from sessions.
 func (t *Topic) handleClientMsg(msg *ClientComMessage) {
+	started := time.Now()
 	kind := "other"
 	if t.cat == types.TopicCatP2P {
 		kind = "p2p"
 	} else if t.cat == types.TopicCatGrp {
 		kind = "group"
 	}
+	operation := "other"
+	if msg.Pub != nil {
+		operation = "publish"
+	} else if msg.Note != nil {
+		operation = "note"
+	}
+	defer immetrics.ObserveTopicProcess(kind, t.metricRole(), operation, started)
 	immetrics.ObserveTopicQueueWait(kind, msg.queuedAt)
 	msg.queuedAt = time.Time{}
 	if msg.TenantID.IsZero() {
@@ -1958,9 +1966,18 @@ func (t *Topic) handlePresence(msg *ServerComMessage) {
 
 // broadcastToSessions writes message to attached sessions.
 func (t *Topic) broadcastToSessions(msg *ServerComMessage) {
+	started := time.Now()
+	kind := t.metricKind()
+	role := t.metricRole()
+	websocketTargets, multiplexTargets, otherTargets := 0, 0, 0
+	defer func() {
+		immetrics.ObserveTopicBroadcast(kind, role, started, websocketTargets, multiplexTargets, otherTargets)
+	}()
 	msg.TenantID = t.tenantID
 	// List of sessions to be dropped.
 	var dropSessions []*Session
+	var sharedPayload []byte
+	sharedPayloadReady := false
 	// Broadcast the message. Only {data}, {pres}, {info} are broadcastable.
 	// {meta} and {ctrl} are sent to the session only
 	for sess, pssd := range t.sessions {
@@ -2027,7 +2044,41 @@ func (t *Topic) broadcastToSessions(msg *ServerComMessage) {
 			}
 		}
 
-		// Make a copy of msg since messages sent to sessions differ.
+		if sess.proto == WEBSOCK {
+			websocketTargets++
+		} else if sess.isMultiplex() {
+			multiplexTargets++
+		} else {
+			otherTargets++
+		}
+
+		// Ordinary group messages are identical for every local WebSocket subscriber.
+		// Serialize once and reuse immutable bytes to avoid O(recipients) JSON work.
+		if t.canShareSerializedBroadcast(msg, sess, pssd) {
+			if !sharedPayloadReady {
+				var err error
+				sharedPayload, err = json.Marshal(msg)
+				if err != nil {
+					immetrics.IncTopicSharedSerialization(kind, "error")
+					logs.Warn.Printf("topic[%s]: shared broadcast serialization failed: %v", t.name, err)
+					sharedPayload = nil
+				} else {
+					immetrics.IncTopicSharedSerialization(kind, "serialized")
+				}
+				sharedPayloadReady = true
+			}
+			if sharedPayload != nil {
+				if sess.queueOutSharedBytes(sharedPayload) {
+					continue
+				}
+				immetrics.IncTopicSharedSerialization(kind, "full")
+				logs.Warn.Printf("topic[%s]: connection stuck, detaching - %s", t.name, sess.sid)
+				dropSessions = append(dropSessions, sess)
+				continue
+			}
+		}
+
+		// Make a copy of msg since messages sent to sessions may differ.
 		msgCopy := msg.copy()
 		// Topic name may be different depending on the user to which the `sess` belongs.
 		t.prepareBroadcastableMessage(msgCopy, pssd.uid, pssd.isChanSub)
@@ -2046,6 +2097,48 @@ func (t *Topic) broadcastToSessions(msg *ServerComMessage) {
 	}
 }
 
+func (t *Topic) metricKind() string {
+	switch t.cat {
+	case types.TopicCatGrp:
+		return "group"
+	case types.TopicCatP2P:
+		return "p2p"
+	default:
+		return "other"
+	}
+}
+
+func (t *Topic) metricRole() string {
+	if t.isProxy {
+		return "proxy"
+	}
+	return "master"
+}
+
+func serverMessageType(msg *ServerComMessage) string {
+	switch {
+	case msg == nil:
+		return "unknown"
+	case msg.Data != nil:
+		return "data"
+	case msg.Ctrl != nil:
+		return "ctrl"
+	case msg.Pres != nil:
+		return "pres"
+	case msg.Info != nil:
+		return "info"
+	case msg.Meta != nil:
+		return "meta"
+	default:
+		return "unknown"
+	}
+}
+
+func (t *Topic) canShareSerializedBroadcast(msg *ServerComMessage, sess *Session, pssd perSessionData) bool {
+	return t.cat == types.TopicCatGrp && !t.isChan && sess.proto == WEBSOCK && !pssd.isChanSub &&
+		(msg.Data != nil || msg.Pres != nil || msg.Info != nil)
+}
+
 // subscriptionReply generates a response to a subscription request
 func (t *Topic) subscriptionReply(asChan bool, msg *ClientComMessage) error {
 	// The topic is already initialized by the Hub

server/topic_proxy.go

@@ -213,6 +213,7 @@ func (t *Topic) handleProxyLeaveRequest(msg *ClientComMessage, killTimer *time.T
 
 // proxyMasterResponse at proxy topic processes a master topic response to an earlier request.
 func (t *Topic) proxyMasterResponse(msg *ClusterResp, killTimer *time.Timer) {
+	immetrics.ObserveClusterProxyQueueWait(serverMessageType(msg.SrvMsg), msg.proxyQueuedAt)
 	// Kills topic after a period of inactivity.
 	keepAlive := idleProxyTopicTimeout
 

server/topic_test.go

@@ -4334,3 +4334,75 @@ func TestMain(m *testing.M) {
 	globals.maxSubscriberCount = 1_000_000_000
 	os.Exit(m.Run())
 }
+
+func TestGroupBroadcastReusesSerializedWebSocketPayload(t *testing.T) {
+	uid1, uid2, uid3 := types.Uid(1), types.Uid(2), types.Uid(3)
+	sess1 := &Session{sid: "ws-1", proto: WEBSOCK, send: make(chan any, 2)}
+	sess2 := &Session{sid: "ws-2", proto: WEBSOCK, send: make(chan any, 2)}
+	sessNoRead := &Session{sid: "ws-no-read", proto: WEBSOCK, send: make(chan any, 2)}
+	topic := &Topic{
+		tenantID:  1,
+		name:      "grpShared",
+		xoriginal: "grpShared",
+		cat:       types.TopicCatGrp,
+		sessions: map[*Session]perSessionData{
+			sess1:      {uid: uid1},
+			sess2:      {uid: uid2},
+			sessNoRead: {uid: uid3},
+		},
+		perUser: map[types.Uid]perUserData{
+			uid1: {modeWant: types.ModeCFull, modeGiven: types.ModeCFull},
+			uid2: {modeWant: types.ModeCFull, modeGiven: types.ModeCFull},
+			uid3: {modeWant: types.ModeNone, modeGiven: types.ModeNone},
+		},
+	}
+	msg := &ServerComMessage{Data: &MsgServerData{
+		Topic: "grpShared", From: uid1.UserId(), SeqId: 42, Timestamp: time.Now(), Content: "hello",
+	}}
+
+	topic.broadcastToSessions(msg)
+
+	payload1, ok := (<-sess1.send).(sharedServerMessage)
+	if !ok {
+		t.Fatal("first WebSocket did not receive a serialized payload")
+	}
+	payload2, ok := (<-sess2.send).(sharedServerMessage)
+	if !ok {
+		t.Fatal("second WebSocket did not receive a serialized payload")
+	}
+	if len(payload1) == 0 || len(payload2) == 0 || &payload1[0] != &payload2[0] {
+		t.Fatal("group WebSocket recipients did not reuse the same serialized payload")
+	}
+	var decoded ServerComMessage
+	if err := json.Unmarshal(payload1, &decoded); err != nil {
+		t.Fatalf("invalid shared payload: %v", err)
+	}
+	if decoded.Data == nil || decoded.Data.Topic != "grpShared" || decoded.Data.SeqId != 42 || decoded.Data.Content != "hello" {
+		t.Fatalf("unexpected shared payload: %#v", decoded.Data)
+	}
+	if len(sessNoRead.send) != 0 {
+		t.Fatal("non-reader received a group broadcast")
+	}
+}
+
+func TestSharedBroadcastEligibility(t *testing.T) {
+	sess := &Session{proto: WEBSOCK}
+	msg := &ServerComMessage{Data: &MsgServerData{Topic: "grpShared"}}
+	for _, tc := range []struct {
+		name  string
+		topic Topic
+		pssd  perSessionData
+		want  bool
+	}{
+		{name: "ordinary group", topic: Topic{cat: types.TopicCatGrp}, want: true},
+		{name: "channel", topic: Topic{cat: types.TopicCatGrp, isChan: true}},
+		{name: "channel subscription", topic: Topic{cat: types.TopicCatGrp}, pssd: perSessionData{isChanSub: true}},
+		{name: "p2p", topic: Topic{cat: types.TopicCatP2P}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := tc.topic.canShareSerializedBroadcast(msg, sess, tc.pssd); got != tc.want {
+				t.Fatalf("eligibility=%v want=%v", got, tc.want)
+			}
+		})
+	}
+}