fix(server): harden user cache update queue under load

server/push.go

@@ -389,9 +389,6 @@ func sendPush(rcpt *push.Receipt) {
 	}
 
 	if len(local.PushRcpt.To) > 0 || local.PushRcpt.Channel != "" {
-		select {
+		enqueueUserUpdate(local, "push", false)
-		case globals.usersUpdate <- local:
-		default:
-		}
 	}
 }

server/user.go

@@ -6,12 +6,14 @@ import (
 	"errors"
 	"math/rand"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"slices"
 
 	"github.com/tinode/chat/server/auth"
 	"github.com/tinode/chat/server/logs"
+	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/push"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
@@ -23,8 +25,15 @@ const (
 	unreadUpdateIOPending = -1
 	// Counter initialization error.
 	unreadUpdateError = -2
+
+	usersUpdateQueueCapacity       = 8192
+	usersUpdateEnqueueTimeout      = 100 * time.Millisecond
+	usersUpdateQueueFullLogEvery   = 5 * time.Second
+	usersUpdateInternalMetricQueue = "user_cache_update"
 )
 
+var usersUpdateQueueFullLastLog atomic.Int64
+
 // Process request for a new account.
 func replyCreateUser(s *Session, msg *ClientComMessage, rec *auth.Rec) {
 	// The session cannot authenticate with the new account because  it's already authenticated.
@@ -1325,7 +1334,9 @@ func (pq *pendingReceiptsQueue) fix(index int) {
 
 // Initialize users cache.
 func usersInit() {
-	globals.usersUpdate = make(chan *UserCacheReq, 1024)
+	globals.usersUpdate = make(chan *UserCacheReq, usersUpdateQueueCapacity)
+	immetrics.SetInternalQueueCurrent(usersUpdateInternalMetricQueue, 0)
+	immetrics.ObserveInternalQueueEnqueue(usersUpdateInternalMetricQueue, 0, usersUpdateQueueCapacity, true)
 
 	go userUpdater()
 }
@@ -1347,10 +1358,7 @@ func usersUpdateUnread(tenantID types.TenantID, uid types.Uid, val int64, inc bo
 		// Send request to remote node which owns the user.
 		globals.cluster.routeUserReq(upd)
 	} else {
-		select {
+		enqueueUserUpdate(upd, "unread", false)
-		case globals.usersUpdate <- upd:
-		default:
-		}
 	}
 }
 
@@ -1368,10 +1376,7 @@ func usersRegisterUser(tenantID types.TenantID, uid types.Uid, add bool) {
 		// Send request to remote node which owns the user.
 		globals.cluster.routeUserReq(upd)
 	} else {
-		select {
+		enqueueUserUpdate(upd, "user", true)
-		case globals.usersUpdate <- upd:
-		default:
-		}
 	}
 }
 
@@ -1383,10 +1388,7 @@ func usersRemoveUser(tenantID types.TenantID, uid types.Uid) {
 
 	upd := &UserCacheReq{TenantID: tenantID, UserId: uid, Gone: true}
 	if !globals.cluster.isRemoteTopic(types.TopicKey{TenantID: tenantID, Topic: uid.UserId()}) {
-		select {
+		enqueueUserUpdate(upd, "remove_user", true)
-		case globals.usersUpdate <- upd:
-		default:
-		}
 	}
 
 	if globals.cluster != nil {
@@ -1431,24 +1433,56 @@ func usersRegisterTopic(t *Topic, add bool) {
 	}
 
 	if len(local.UserIdList) > 0 {
-		select {
+		enqueueUserUpdate(local, "topic", true)
-		case globals.usersUpdate <- local:
-		default:
-			logs.Err.Println("User cache: globals.usersUpdate queue full: ", len(globals.usersUpdate))
-		}
 	}
 }
 
 // usersRequestFromCluster handles requests which came from other cluser nodes.
 func usersRequestFromCluster(req *UserCacheReq) {
-	if globals.usersUpdate == nil {
+	enqueueUserUpdate(req, "cluster", true)
-		return
+}
+
+func enqueueUserUpdate(req *UserCacheReq, kind string, waitOnFull bool) bool {
+	q := globals.usersUpdate
+	if req == nil || q == nil {
+		return false
 	}
 
 	select {
-	case globals.usersUpdate <- req:
+	case q <- req:
+		immetrics.ObserveInternalQueueEnqueue(usersUpdateInternalMetricQueue, len(q), cap(q), true)
+		immetrics.SetInternalQueueCurrent(usersUpdateInternalMetricQueue, len(q))
+		return true
 	default:
 	}
+
+	if waitOnFull {
+		timer := time.NewTimer(usersUpdateEnqueueTimeout)
+		defer timer.Stop()
+		select {
+		case q <- req:
+			immetrics.ObserveInternalQueueEnqueue(usersUpdateInternalMetricQueue, len(q), cap(q), true)
+			immetrics.SetInternalQueueCurrent(usersUpdateInternalMetricQueue, len(q))
+			return true
+		case <-timer.C:
+		}
+	}
+
+	immetrics.ObserveInternalQueueEnqueue(usersUpdateInternalMetricQueue, len(q), cap(q), false)
+	immetrics.SetInternalQueueCurrent(usersUpdateInternalMetricQueue, len(q))
+	logUsersUpdateQueueFull(kind, q)
+	return false
+}
+
+func logUsersUpdateQueueFull(kind string, q chan *UserCacheReq) {
+	now := time.Now().UnixNano()
+	last := usersUpdateQueueFullLastLog.Load()
+	if time.Duration(now-last) < usersUpdateQueueFullLogEvery ||
+		!usersUpdateQueueFullLastLog.CompareAndSwap(last, now) {
+		return
+	}
+	logs.Err.Printf("User cache: usersUpdate queue full kind=%s len=%d cap=%d wait=%s",
+		kind, len(q), cap(q), usersUpdateEnqueueTimeout)
 }
 
 var usersCache map[types.UserKey]userCacheEntry
@@ -1590,6 +1624,7 @@ func userUpdater() {
 				push.Push(rcpt)
 			}
 		case upd := <-globals.usersUpdate:
+			immetrics.SetInternalQueueCurrent(usersUpdateInternalMetricQueue, len(globals.usersUpdate))
 			if globals.shuttingDown {
 				// If shutdown is in progress we don't care to process anything.
 				// ignore all calls.