fix(server): shed overloaded group websocket fanout

server/metrics/metrics.go

@@ -209,6 +209,10 @@ var (
 		Name: "im_topic_shared_serialization_total",
 		Help: "Shared WebSocket serialization events by pod, topic kind, and result.",
 	}, []string{"pod", "kind", "result"})
+	TopicBroadcastDropTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "im_topic_broadcast_drop_total",
+		Help: "Realtime topic broadcasts skipped by pod, topic kind, role, and reason.",
+	}, []string{"pod", "kind", "role", "reason"})
 	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.",
@@ -282,7 +286,7 @@ func init() {
 		ClusterRouteTotal, ClusterProxyResponseTotal, InternalQueueEnqueueTotal, InternalQueueDepth,
 		InternalQueueCapacity, InternalQueueCurrent, MessagePersistDuration, MessagePublishStageDuration, TopicQueueWaitDuration,
 		TopicProcessDuration, TopicBroadcastDuration, TopicBroadcastTargets, TopicSharedSerializationTotal,
-		ClusterM2PQueueWaitDuration, ClusterM2PQueueDepth, ClusterProxyQueueWaitDuration, WSReadDispatchDuration,
+		TopicBroadcastDropTotal, ClusterM2PQueueWaitDuration, ClusterM2PQueueDepth, ClusterProxyQueueWaitDuration, WSReadDispatchDuration,
 		PushDispatchQueueEnqueueTotal, PushDispatchQueueDepth, PushDispatchQueueWaitDuration,
 		PushDispatchDuration, PushRecipientCount,
 		KafkaMessagePersistTotal)
@@ -339,6 +343,10 @@ func IncTopicSharedSerialization(kind, result string) {
 	TopicSharedSerializationTotal.WithLabelValues(metricPodName(), normalizeLabel(kind), normalizeLabel(result)).Inc()
 }
 
+func IncTopicBroadcastDrop(kind, role, reason string) {
+	TopicBroadcastDropTotal.WithLabelValues(metricPodName(), normalizeLabel(kind), normalizeLabel(role), normalizeLabel(reason)).Inc()
+}
+
 func ObserveClusterM2PEnqueue(peer string, depth int) {
 	ClusterM2PQueueDepth.WithLabelValues(metricPodName(), normalizeLabel(peer)).Observe(float64(depth))
 }

server/monitoring_http_test.go

@@ -101,6 +101,7 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) {
 	immetrics.ObserveTopicProcess("group", "master", "publish", time.Now())
 	immetrics.ObserveTopicBroadcast("group", "master", time.Now(), 1, 0, 0)
 	immetrics.IncTopicSharedSerialization("group", "serialized")
+	immetrics.IncTopicBroadcastDrop("group", "master", "ws_backpressure")
 	immetrics.ObserveClusterM2PEnqueue("peer", 1)
 	immetrics.ObserveClusterM2PDequeue("peer", time.Now())
 	immetrics.ObserveClusterProxyQueueWait("data", time.Now())
@@ -126,6 +127,7 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) {
 		"im_internal_queue_enqueue_total", "im_internal_queue_depth", "im_internal_queue_capacity", "im_internal_queue_current",
 		"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_topic_broadcast_drop_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",

server/session.go

@@ -35,6 +35,11 @@ import (
 // Maximum number of queued messages before session is considered stale and dropped.
 const sendQueueLimit = 128
 
+const (
+	lossyBroadcastQueueLimit = sendQueueLimit / 2
+	lossyBroadcastMaxWait    = 2 * time.Second
+)
+
 // Multiplex sessions may receive a short burst of presence updates while many
 // remote connections are closing. Give the cluster drain a bounded chance to
 // free one slot before treating the proxy as stuck.
@@ -547,6 +552,38 @@ func (s *Session) queueOutSharedBytes(data []byte) bool {
 	return true
 }
 
+func (s *Session) queueOutLossySharedBytes(data []byte) (queued bool, skipped bool) {
+	if s == nil || atomic.LoadInt32(&s.terminating) > 0 {
+		return true, false
+	}
+	if s.proto != WEBSOCK {
+		return s.queueOutSharedBytes(data), false
+	}
+
+	s.wsQueueMu.Lock()
+	depth := len(s.send)
+	stale := false
+	if s.wsQueueHead < len(s.wsQueueTimes) {
+		stale = time.Since(s.wsQueueTimes[s.wsQueueHead]) > lossyBroadcastMaxWait
+	}
+	if depth >= lossyBroadcastQueueLimit || stale {
+		immetrics.ObserveWSSendQueueEnqueue(depth, cap(s.send), false)
+		s.wsQueueMu.Unlock()
+		return true, true
+	}
+	select {
+	case s.send <- sharedServerMessage(data):
+		s.wsQueueTimes = append(s.wsQueueTimes, time.Now())
+		immetrics.ObserveWSSendQueueEnqueue(len(s.send), cap(s.send), true)
+		s.wsQueueMu.Unlock()
+		return true, false
+	default:
+		immetrics.ObserveWSSendQueueEnqueue(len(s.send), cap(s.send), false)
+		s.wsQueueMu.Unlock()
+		return true, true
+	}
+}
+
 func (s *Session) maybeScheduleClusterWriteLoop() {
 	if s.multi != nil {
 		s.multi.scheduleClusterWriteLoop()

server/topic.go

@@ -2068,7 +2068,15 @@ func (t *Topic) broadcastToSessions(msg *ServerComMessage) {
 				sharedPayloadReady = true
 			}
 			if sharedPayload != nil {
-				if sess.queueOutSharedBytes(sharedPayload) {
+				if msg.Data != nil {
+					queued, skipped := sess.queueOutLossySharedBytes(sharedPayload)
+					if skipped {
+						immetrics.IncTopicBroadcastDrop(kind, role, "ws_backpressure")
+					}
+					if queued {
+						continue
+					}
+				} else if sess.queueOutSharedBytes(sharedPayload) {
 					continue
 				}
 				immetrics.IncTopicSharedSerialization(kind, "full")