diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 87aaa80..107c71a 100644 --- a/server/metrics/metrics.go +++ b/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)) } diff --git a/server/monitoring_http_test.go b/server/monitoring_http_test.go index 9dab5a2..47932fd 100644 --- a/server/monitoring_http_test.go +++ b/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", diff --git a/server/session.go b/server/session.go index 298f138..9035f90 100644 --- a/server/session.go +++ b/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() diff --git a/server/topic.go b/server/topic.go index 9a4ae0e..172de61 100644 --- a/server/topic.go +++ b/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")