fix: absorb multiplex close bursts

server/cluster.go

@@ -1401,8 +1401,10 @@ func (sess *Session) clusterWriteLoop(forTopic string) {
 			// The topic may retain this multiplexing session briefly while its
 			// unsubscribe request is processed. Reject writes to the stale session
 			// instead of filling a send queue which no longer has a drain owner.
+			sess.clusterQueueMu.Lock()
 			atomic.StoreInt32(&sess.terminating, 1)
 			sess.purgeChannels()
+			sess.clusterQueueMu.Unlock()
 			sess.closeRPC()
 			globals.sessionStore.Delete(sess)
 			sess.inflightReqs = nil

server/cluster_order_test.go

@@ -112,3 +112,54 @@ func TestClusterWriteLoopRejectsWritesAfterTermination(t *testing.T) {
 		t.Fatalf("terminated multiplexing session retained %d outbound messages", len(sess.send))
 	}
 }
+
+func TestMultiplexEnqueueWaitsForDrain(t *testing.T) {
+	sess := &Session{proto: MULTIPLEX, send: make(chan any, 1)}
+	sess.send <- "existing"
+
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		<-sess.send
+	}()
+
+	payload := &ServerComMessage{TenantID: types.TenantID(1)}
+	if !sess.enqueueMultiplexOutbound(payload, 200*time.Millisecond) {
+		t.Fatal("multiplex enqueue failed while the drain was making progress")
+	}
+	if got := <-sess.send; got != payload {
+		t.Fatalf("unexpected queued payload: %#v", got)
+	}
+}
+
+func TestMultiplexEnqueueTimesOutWithoutDrain(t *testing.T) {
+	sess := &Session{proto: MULTIPLEX, send: make(chan any, 1)}
+	sess.send <- "existing"
+
+	started := time.Now()
+	if sess.enqueueMultiplexOutbound("next", 20*time.Millisecond) {
+		t.Fatal("multiplex enqueue succeeded without queue capacity")
+	}
+	if elapsed := time.Since(started); elapsed < 15*time.Millisecond || elapsed > 200*time.Millisecond {
+		t.Fatalf("multiplex enqueue timeout outside expected bounds: %s", elapsed)
+	}
+}
+
+func TestMultiplexEnqueueStopsWhenSessionTerminates(t *testing.T) {
+	sess := &Session{proto: MULTIPLEX, send: make(chan any, 1)}
+	sess.send <- "existing"
+
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		sess.clusterQueueMu.Lock()
+		atomic.StoreInt32(&sess.terminating, 1)
+		<-sess.send
+		sess.clusterQueueMu.Unlock()
+	}()
+
+	if !sess.enqueueMultiplexOutbound("next", 200*time.Millisecond) {
+		t.Fatal("terminating multiplex session should discard the pending write")
+	}
+	if len(sess.send) != 0 {
+		t.Fatalf("terminating multiplex session retained %d outbound messages", len(sess.send))
+	}
+}

server/session.go

@@ -35,6 +35,11 @@ import (
 // Maximum number of queued messages before session is considered stale and dropped.
 const sendQueueLimit = 128
 
+// 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.
+const multiplexEnqueueWait = 250 * time.Millisecond
+
 // Time given to a background session to terminate to avoid tiggering presence notifications.
 // If session terminates (or unsubscribes from topic) in this time frame notifications are not sent at all.
 const deferredNotificationsTimeout = time.Second * 5
@@ -171,6 +176,7 @@ type Session struct {
 	// Ensures that one multiplexing session is drained by at most one cluster
 	// write task at a time, preserving per-topic response order.
 	clusterWriteScheduled atomic.Bool
+	clusterQueueMu        sync.Mutex
 }
 
 // Subscription is a mapper of sessions to topics.
@@ -190,6 +196,9 @@ type Subscription struct {
 }
 
 func (s *Session) enqueueOutbound(payload any) bool {
+	if s.isMultiplex() {
+		return s.enqueueMultiplexOutbound(payload, multiplexEnqueueWait)
+	}
 	if s.proto != WEBSOCK {
 		select {
 		case s.send <- payload:
@@ -212,6 +221,42 @@ func (s *Session) enqueueOutbound(payload any) bool {
 	}
 }
 
+func (s *Session) enqueueMultiplexOutbound(payload any, wait time.Duration) bool {
+	deadline := time.NewTimer(wait)
+	defer deadline.Stop()
+	retry := time.NewTicker(time.Millisecond)
+	defer retry.Stop()
+
+	drainScheduled := false
+	for {
+		s.clusterQueueMu.Lock()
+		if atomic.LoadInt32(&s.terminating) > 0 {
+			s.clusterQueueMu.Unlock()
+			return true
+		}
+		select {
+		case s.send <- payload:
+			s.clusterQueueMu.Unlock()
+			return true
+		default:
+			s.clusterQueueMu.Unlock()
+		}
+
+		if !drainScheduled {
+			s.scheduleClusterWriteLoop()
+			drainScheduled = true
+		}
+		select {
+		case <-retry.C:
+		case <-deadline.C:
+			s.clusterQueueMu.Lock()
+			terminating := atomic.LoadInt32(&s.terminating) > 0
+			s.clusterQueueMu.Unlock()
+			return terminating
+		}
+	}
+}
+
 func (s *Session) observeOutboundDequeue(purged bool) {
 	if s.proto != WEBSOCK {
 		return