[IMPROVED] Reduce lock contention in accountDisconnectEvent

Protect event ID generation with a dedicated mutex. This
allows accountDisconnectEvent to check whether events are
enabled under the server read lock, then generate the event
ID without holding the server lock. This avoids serializing
concurrent disconnects on the server write lock.
Apply the same pattern to all methods that generate event IDs.

Signed-off-by: Daniele Sciascia <daniele@nats.io>

Author: sciascid

locksordering.txt

@@ -59,3 +59,9 @@ mirrors/sources, info requests, etc.) are sequentially consistent and may observ
 an inflight batch. It must be acquired before the stream lock, never while holding it.
 
     isolateMu -> stream
+
+The "eventIdsMu" lock protects the server's event ID generator. It may be acquired
+while holding the Server lock, but the Server lock must not be acquired while
+holding eventIdsMu.
+
+    Server -> eventIdsMu

server/events.go

@@ -2561,21 +2561,23 @@ func (s *Server) accConnsUpdate(a *Account) {
 	s.sendAccConnsUpdate(a, fmt.Sprintf(accConnsEventSubjOld, a.Name), fmt.Sprintf(accConnsEventSubjNew, a.Name))
 }
 
-// server lock should be held
 func (s *Server) nextEventID() string {
-	return s.eventIds.Next()
+	s.eventIdsMu.Lock()
+	id := s.eventIds.Next()
+	s.eventIdsMu.Unlock()
+	return id
 }
 
 // accountConnectEvent will send an account client connect event if there is interest.
 // This is a billing event.
 func (s *Server) accountConnectEvent(c *client) {
-	s.mu.Lock()
-	if !s.eventsEnabled() {
-		s.mu.Unlock()
+	s.mu.RLock()
+	eventsEnabled := s.eventsEnabled()
+	s.mu.RUnlock()
+	if !eventsEnabled {
 		return
 	}
 	eid := s.nextEventID()
-	s.mu.Unlock()
 
 	c.mu.Lock()
 	if c.acc == nil {
@@ -2616,13 +2618,13 @@ func (s *Server) accountConnectEvent(c *client) {
 // accountDisconnectEvent will send an account client disconnect event if there is interest.
 // This is a billing event.
 func (s *Server) accountDisconnectEvent(c *client, now time.Time, reason string) {
-	s.mu.Lock()
-	if !s.eventsEnabled() {
-		s.mu.Unlock()
+	s.mu.RLock()
+	eventsEnabled := s.eventsEnabled()
+	s.mu.RUnlock()
+	if !eventsEnabled {
 		return
 	}
 	eid := s.nextEventID()
-	s.mu.Unlock()
 
 	c.mu.Lock()
 
@@ -2678,13 +2680,13 @@ func (s *Server) accountDisconnectEvent(c *client, now time.Time, reason string)
 
 // This is the system level event sent to the system account for operators.
 func (s *Server) sendAuthErrorEvent(c *client, reason string) {
-	s.mu.Lock()
-	if !s.eventsEnabled() {
-		s.mu.Unlock()
+	s.mu.RLock()
+	eventsEnabled := s.eventsEnabled()
+	s.mu.RUnlock()
+	if !eventsEnabled {
 		return
 	}
 	eid := s.nextEventID()
-	s.mu.Unlock()
 
 	now := time.Now().UTC()
 	c.mu.Lock()
@@ -2740,13 +2742,13 @@ func (s *Server) sendAccountAuthErrorEvent(c *client, acc *Account, reason strin
 	if acc == nil {
 		return
 	}
-	s.mu.Lock()
-	if !s.eventsEnabled() {
-		s.mu.Unlock()
+	s.mu.RLock()
+	eventsEnabled := s.eventsEnabled()
+	s.mu.RUnlock()
+	if !eventsEnabled {
 		return
 	}
 	eid := s.nextEventID()
-	s.mu.Unlock()
 
 	now := time.Now().UTC()
 	c.mu.Lock()
@@ -3314,9 +3316,11 @@ func (s *Server) wrapChk(f func()) func() {
 // sendOCSPPeerRejectEvent sends a system level event to system account when a peer connection is
 // rejected due to OCSP invalid status of its trust chain(s).
 func (s *Server) sendOCSPPeerRejectEvent(kind string, peer *x509.Certificate, reason string) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if !s.eventsEnabled() {
+	s.mu.RLock()
+	eventsEnabled := s.eventsEnabled()
+	serverID := s.info.ID
+	s.mu.RUnlock()
+	if !eventsEnabled {
 		return
 	}
 	if peer == nil {
@@ -3340,16 +3344,18 @@ func (s *Server) sendOCSPPeerRejectEvent(kind string, peer *x509.Certificate, re
 		},
 		Reason: reason,
 	}
-	subj := fmt.Sprintf(ocspPeerRejectEventSubj, s.info.ID)
-	s.sendInternalMsg(subj, _EMPTY_, &m.Server, &m)
+	subj := fmt.Sprintf(ocspPeerRejectEventSubj, serverID)
+	s.sendInternalMsgLocked(subj, _EMPTY_, &m.Server, &m)
 }
 
 // sendOCSPPeerChainlinkInvalidEvent sends a system level event to system account when a link in a peer's trust chain
 // is OCSP invalid.
 func (s *Server) sendOCSPPeerChainlinkInvalidEvent(peer *x509.Certificate, link *x509.Certificate, reason string) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if !s.eventsEnabled() {
+	s.mu.RLock()
+	eventsEnabled := s.eventsEnabled()
+	serverID := s.info.ID
+	s.mu.RUnlock()
+	if !eventsEnabled {
 		return
 	}
 	if peer == nil || link == nil {
@@ -3378,6 +3384,6 @@ func (s *Server) sendOCSPPeerChainlinkInvalidEvent(peer *x509.Certificate, link
 		},
 		Reason: reason,
 	}
-	subj := fmt.Sprintf(ocspPeerChainlinkInvalidEventSubj, s.info.ID)
-	s.sendInternalMsg(subj, _EMPTY_, &m.Server, &m)
+	subj := fmt.Sprintf(ocspPeerChainlinkInvalidEventSubj, serverID)
+	s.sendInternalMsgLocked(subj, _EMPTY_, &m.Server, &m)
 }

server/server.go

@@ -309,7 +309,8 @@ type Server struct {
 	}
 
 	// For eventIDs
-	eventIds *nuid.NUID
+	eventIdsMu sync.Mutex
+	eventIds   *nuid.NUID
 
 	// Websocket structure
 	websocket srvWebsocket