build(deps): bump github.com/nats-io/nats-server/v2 (#762)

Bumps [github.com/nats-io/nats-server/v2](https://github.com/nats-io/nats-server) from 2.11.2 to 2.11.3.
- [Release notes](https://github.com/nats-io/nats-server/releases)
- [Changelog](https://github.com/nats-io/nats-server/blob/main/.goreleaser.yml)
- [Commits](https://github.com/nats-io/nats-server/compare/v2.11.2...v2.11.3)

---
updated-dependencies:
- dependency-name: github.com/nats-io/nats-server/v2
  dependency-version: 2.11.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

vendor/github.com/nats-io/nats-server/v2/server/certstore/certstore_windows.go

@@ -593,7 +593,7 @@ func winSignECDSA(kh uintptr, digest []byte) ([]byte, error) {
 		return nil, ErrStoreECDSASigningError
 	}
 
-	return winPackECDSASigValue(bytes.NewReader(buf[:size]), len(digest))
+	return winPackECDSASigValue(bytes.NewReader(buf[:size]), int(size/2))
 }
 
 func winPackECDSASigValue(r io.Reader, digestLength int) ([]byte, error) {

vendor/github.com/nats-io/nats-server/v2/server/const.go

@@ -58,7 +58,7 @@ func init() {
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.11.2"
+	VERSION = "2.11.3"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original

vendor/github.com/nats-io/nats-server/v2/server/consumer.go

@@ -2983,13 +2983,14 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		}
 	}
 
-	// If we are replicated, we need to pull certain data from our store.
-	if rg != nil && rg.node != nil && o.store != nil {
+	// We always need to pull certain data from our store.
+	if o.store != nil {
 		state, err := o.store.BorrowState()
 		if err != nil {
 			o.mu.Unlock()
 			return nil
 		}
+
 		// If we are the leader we could have o.sseq that is skipped ahead.
 		// To maintain consistency in reporting (e.g. jsz) we always take the state for our delivered/ackfloor stream sequence.
 		// Only use skipped ahead o.sseq if we're a new consumer and have not yet replicated this state yet.

vendor/github.com/nats-io/nats-server/v2/server/gsl/gsl.go

@@ -16,6 +16,7 @@ package gsl
 import (
 	"errors"
 	"sync"
+	"unsafe"
 
 	"github.com/nats-io/nats-server/v2/server/stree"
 )
@@ -479,16 +480,6 @@ func IntersectStree[T1 any, T2 comparable](st *stree.SubjectTree[T1], sl *Generi
 }
 
 func intersectStree[T1 any, T2 comparable](st *stree.SubjectTree[T1], r *level[T2], subj []byte, cb func(subj []byte, entry *T1)) {
-	if r.numNodes() == 0 {
-		// For wildcards we can't avoid Match, but if it's a literal subject at
-		// this point, using Find is considerably cheaper.
-		if subjectHasWildcard(string(subj)) {
-			st.Match(subj, cb)
-		} else if e, ok := st.Find(subj); ok {
-			cb(subj, e)
-		}
-		return
-	}
 	nsubj := subj
 	if len(nsubj) > 0 {
 		nsubj = append(subj, '.')
@@ -504,15 +495,28 @@ func intersectStree[T1 any, T2 comparable](st *stree.SubjectTree[T1], r *level[T
 		// check whether there's interest at this level (without triggering dupes) and
 		// match if so.
 		nsubj := append(nsubj, '*')
-		if len(r.pwc.subs) > 0 && r.pwc.next != nil && r.pwc.next.numNodes() > 0 {
+		if len(r.pwc.subs) > 0 {
 			st.Match(nsubj, cb)
 		}
-		intersectStree(st, r.pwc.next, nsubj, cb)
-	case r.numNodes() > 0:
+		if r.pwc.next != nil && r.pwc.next.numNodes() > 0 {
+			intersectStree(st, r.pwc.next, nsubj, cb)
+		}
+	default:
 		// Normal node with subject literals, keep iterating.
 		for t, n := range r.nodes {
 			nsubj := append(nsubj, t...)
-			intersectStree(st, n.next, nsubj, cb)
+			if len(n.subs) > 0 {
+				if subjectHasWildcard(bytesToString(nsubj)) {
+					st.Match(nsubj, cb)
+				} else {
+					if e, ok := st.Find(nsubj); ok {
+						cb(nsubj, e)
+					}
+				}
+			}
+			if n.next != nil && n.next.numNodes() > 0 {
+				intersectStree(st, n.next, nsubj, cb)
+			}
 		}
 	}
 }
@@ -530,3 +534,13 @@ func subjectHasWildcard(subject string) bool {
 	}
 	return false
 }
+
+// Note this will avoid a copy of the data used for the string, but it will also reference the existing slice's data pointer.
+// So this should be used sparingly when we know the encompassing byte slice's lifetime is the same.
+func bytesToString(b []byte) string {
+	if len(b) == 0 {
+		return _EMPTY_
+	}
+	p := unsafe.SliceData(b)
+	return unsafe.String(p, len(b))
+}

vendor/github.com/nats-io/nats-server/v2/server/monitor.go

@@ -340,7 +340,17 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 
 	// Search by individual CID.
 	if cid > 0 {
-		if state == ConnClosed || state == ConnAll {
+		// Let's first check if user also selects on ConnOpen or ConnAll
+		// and look for opened connections.
+		if state == ConnOpen || state == ConnAll {
+			if client := s.clients[cid]; client != nil {
+				openClients = append(openClients, client)
+				closedClients = nil
+			}
+		}
+		// If we did not find, and the user selected for ConnClosed or ConnAll,
+		// look for closed connections.
+		if len(openClients) == 0 && (state == ConnClosed || state == ConnAll) {
 			copyClosed := closedClients
 			closedClients = nil
 			for _, cc := range copyClosed {
@@ -349,11 +359,6 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 					break
 				}
 			}
-		} else if state == ConnOpen || state == ConnAll {
-			client := s.clients[cid]
-			if client != nil {
-				openClients = append(openClients, client)
-			}
 		}
 	} else {
 		// Gather all open clients.

vendor/github.com/nats-io/nats-server/v2/server/mqtt.go

@@ -189,6 +189,7 @@ const (
 	mqttProcessSubTooLong       = 100 * time.Millisecond
 	mqttDefaultRetainedCacheTTL = 2 * time.Minute
 	mqttRetainedTransferTimeout = 10 * time.Second
+	mqttDefaultJSAPITimeout     = 5 * time.Second
 )
 
 const (
@@ -209,30 +210,30 @@ var (
 	mqttOldProtoName     = []byte("MQIsdp")
 	mqttSessJailDur      = mqttSessFlappingJailDur
 	mqttFlapCleanItvl    = mqttSessFlappingCleanupInterval
-	mqttJSAPITimeout     = 4 * time.Second
 	mqttRetainedCacheTTL = mqttDefaultRetainedCacheTTL
 )
 
 var (
-	errMQTTNotWebsocketPort         = errors.New("MQTT clients over websocket must connect to the Websocket port, not the MQTT port")
-	errMQTTTopicFilterCannotBeEmpty = errors.New("topic filter cannot be empty")
-	errMQTTMalformedVarInt          = errors.New("malformed variable int")
-	errMQTTSecondConnectPacket      = errors.New("received a second CONNECT packet")
-	errMQTTServerNameMustBeSet      = errors.New("mqtt requires server name to be explicitly set")
-	errMQTTUserMixWithUsersNKeys    = errors.New("mqtt authentication username not compatible with presence of users/nkeys")
-	errMQTTTokenMixWIthUsersNKeys   = errors.New("mqtt authentication token not compatible with presence of users/nkeys")
-	errMQTTAckWaitMustBePositive    = errors.New("ack wait must be a positive value")
-	errMQTTStandaloneNeedsJetStream = errors.New("mqtt requires JetStream to be enabled if running in standalone mode")
-	errMQTTConnFlagReserved         = errors.New("connect flags reserved bit not set to 0")
-	errMQTTWillAndRetainFlag        = errors.New("if Will flag is set to 0, Will Retain flag must be 0 too")
-	errMQTTPasswordFlagAndNoUser    = errors.New("password flag set but username flag is not")
-	errMQTTCIDEmptyNeedsCleanFlag   = errors.New("when client ID is empty, clean session flag must be set to 1")
-	errMQTTEmptyWillTopic           = errors.New("empty Will topic not allowed")
-	errMQTTEmptyUsername            = errors.New("empty user name not allowed")
-	errMQTTTopicIsEmpty             = errors.New("topic cannot be empty")
-	errMQTTPacketIdentifierIsZero   = errors.New("packet identifier cannot be 0")
-	errMQTTUnsupportedCharacters    = errors.New("character ' ' not supported for MQTT topics")
-	errMQTTInvalidSession           = errors.New("invalid MQTT session")
+	errMQTTNotWebsocketPort           = errors.New("MQTT clients over websocket must connect to the Websocket port, not the MQTT port")
+	errMQTTTopicFilterCannotBeEmpty   = errors.New("topic filter cannot be empty")
+	errMQTTMalformedVarInt            = errors.New("malformed variable int")
+	errMQTTSecondConnectPacket        = errors.New("received a second CONNECT packet")
+	errMQTTServerNameMustBeSet        = errors.New("mqtt requires server name to be explicitly set")
+	errMQTTUserMixWithUsersNKeys      = errors.New("mqtt authentication username not compatible with presence of users/nkeys")
+	errMQTTTokenMixWIthUsersNKeys     = errors.New("mqtt authentication token not compatible with presence of users/nkeys")
+	errMQTTAckWaitMustBePositive      = errors.New("ack wait must be a positive value")
+	errMQTTJSAPITimeoutMustBePositive = errors.New("JS API timeout must be a positive value")
+	errMQTTStandaloneNeedsJetStream   = errors.New("mqtt requires JetStream to be enabled if running in standalone mode")
+	errMQTTConnFlagReserved           = errors.New("connect flags reserved bit not set to 0")
+	errMQTTWillAndRetainFlag          = errors.New("if Will flag is set to 0, Will Retain flag must be 0 too")
+	errMQTTPasswordFlagAndNoUser      = errors.New("password flag set but username flag is not")
+	errMQTTCIDEmptyNeedsCleanFlag     = errors.New("when client ID is empty, clean session flag must be set to 1")
+	errMQTTEmptyWillTopic             = errors.New("empty Will topic not allowed")
+	errMQTTEmptyUsername              = errors.New("empty user name not allowed")
+	errMQTTTopicIsEmpty               = errors.New("topic cannot be empty")
+	errMQTTPacketIdentifierIsZero     = errors.New("packet identifier cannot be 0")
+	errMQTTUnsupportedCharacters      = errors.New("character ' ' not supported for MQTT topics")
+	errMQTTInvalidSession             = errors.New("invalid MQTT session")
 )
 
 type srvMQTT struct {
@@ -281,6 +282,7 @@ type mqttJSA struct {
 	quitCh    chan struct{}
 	domain    string // Domain or possibly empty. This is added to session subject.
 	domainSet bool   // covers if domain was set, even to empty
+	timeout   time.Duration
 }
 
 type mqttJSPubMsg struct {
@@ -696,6 +698,9 @@ func validateMQTTOptions(o *Options) error {
 	if mo.AckWait < 0 {
 		return errMQTTAckWaitMustBePositive
 	}
+	if mo.JSAPITimeout < 0 {
+		return errMQTTJSAPITimeoutMustBePositive
+	}
 	// If strictly standalone and there is no JS enabled, then it won't work...
 	// For leafnodes, we could either have remote(s) and it would be ok, or no
 	// remote but accept from a remote side that has "hub" property set, which
@@ -1152,6 +1157,12 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 	c.acc = acc
 
 	id := s.NodeName()
+
+	mqttJSAPITimeout := opts.MQTT.JSAPITimeout
+	if mqttJSAPITimeout == 0 {
+		mqttJSAPITimeout = mqttDefaultJSAPITimeout
+	}
+
 	replicas := opts.MQTT.StreamReplicas
 	if replicas <= 0 {
 		replicas = s.mqttDetermineReplicas()
@@ -1163,12 +1174,13 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 		sessLocked: make(map[string]struct{}),
 		flappers:   make(map[string]int64),
 		jsa: mqttJSA{
-			id:     id,
-			c:      c,
-			rplyr:  mqttJSARepliesPrefix + id + ".",
-			sendq:  newIPQueue[*mqttJSPubMsg](s, qname+"send"),
-			nuid:   nuid.New(),
-			quitCh: quitCh,
+			id:      id,
+			c:       c,
+			rplyr:   mqttJSARepliesPrefix + id + ".",
+			sendq:   newIPQueue[*mqttJSPubMsg](s, qname+"send"),
+			nuid:    nuid.New(),
+			quitCh:  quitCh,
+			timeout: mqttJSAPITimeout,
 		},
 	}
 	if !testDisableRMSCache {
@@ -1546,7 +1558,7 @@ func (s *Server) mqttDetermineReplicas() int {
 //////////////////////////////////////////////////////////////////////////////
 
 func (jsa *mqttJSA) newRequest(kind, subject string, hdr int, msg []byte) (any, error) {
-	return jsa.newRequestEx(kind, subject, _EMPTY_, hdr, msg, mqttJSAPITimeout)
+	return jsa.newRequestEx(kind, subject, _EMPTY_, hdr, msg)
 }
 
 func (jsa *mqttJSA) prefixDomain(subject string) string {
@@ -1559,8 +1571,8 @@ func (jsa *mqttJSA) prefixDomain(subject string) string {
 	return subject
 }
 
-func (jsa *mqttJSA) newRequestEx(kind, subject, cidHash string, hdr int, msg []byte, timeout time.Duration) (any, error) {
-	responses, err := jsa.newRequestExMulti(kind, subject, cidHash, []int{hdr}, [][]byte{msg}, timeout)
+func (jsa *mqttJSA) newRequestEx(kind, subject, cidHash string, hdr int, msg []byte) (any, error) {
+	responses, err := jsa.newRequestExMulti(kind, subject, cidHash, []int{hdr}, [][]byte{msg})
 	if err != nil {
 		return nil, err
 	}
@@ -1578,7 +1590,7 @@ func (jsa *mqttJSA) newRequestEx(kind, subject, cidHash string, hdr int, msg []b
 //
 // Note that each response may represent an error and should be inspected as
 // such by the caller.
-func (jsa *mqttJSA) newRequestExMulti(kind, subject, cidHash string, hdrs []int, msgs [][]byte, timeout time.Duration) ([]*mqttJSAResponse, error) {
+func (jsa *mqttJSA) newRequestExMulti(kind, subject, cidHash string, hdrs []int, msgs [][]byte) ([]*mqttJSAResponse, error) {
 	if len(hdrs) != len(msgs) {
 		return nil, fmt.Errorf("unreachable: invalid number of messages (%d) or header offsets (%d)", len(msgs), len(hdrs))
 	}
@@ -1630,7 +1642,7 @@ func (jsa *mqttJSA) newRequestExMulti(kind, subject, cidHash string, hdrs []int,
 	c := 0
 	responses := make([]*mqttJSAResponse, len(msgs))
 	start := time.Now()
-	t := time.NewTimer(timeout)
+	t := time.NewTimer(jsa.timeout)
 	defer t.Stop()
 	for {
 		select {
@@ -1789,7 +1801,7 @@ func (jsa *mqttJSA) loadLastMsgForMulti(streamName string, subjects []string) ([
 		headerBytes = append(headerBytes, 0)
 	}
 
-	all, err := jsa.newRequestExMulti(mqttJSAMsgLoad, fmt.Sprintf(JSApiMsgGetT, streamName), _EMPTY_, headerBytes, marshaled, mqttJSAPITimeout)
+	all, err := jsa.newRequestExMulti(mqttJSAMsgLoad, fmt.Sprintf(JSApiMsgGetT, streamName), _EMPTY_, headerBytes, marshaled)
 	// all has the same order as subjects, preserve it as we unmarshal
 	responses := make([]*JSApiMsgGetResponse, len(all))
 	for i, v := range all {
@@ -1847,7 +1859,7 @@ func (jsa *mqttJSA) storeSessionMsg(domainTk, cidHash string, hdr int, msg []byt
 
 	// Passing cidHash will add it to the JS reply subject, so that we can use
 	// it in processSessionPersist.
-	smri, err := jsa.newRequestEx(mqttJSASessPersist, subject, cidHash, hdr, msg, mqttJSAPITimeout)
+	smri, err := jsa.newRequestEx(mqttJSASessPersist, subject, cidHash, hdr, msg)
 	if err != nil {
 		return nil, err
 	}
@@ -2982,7 +2994,7 @@ func (as *mqttAccountSessionManager) transferUniqueSessStreamsToMuxed(log *Serve
 	}()
 
 	jsa := &as.jsa
-	sni, err := jsa.newRequestEx(mqttJSAStreamNames, JSApiStreams, _EMPTY_, 0, nil, 5*time.Second)
+	sni, err := jsa.newRequestEx(mqttJSAStreamNames, JSApiStreams, _EMPTY_, 0, nil)
 	if err != nil {
 		log.Errorf("Unable to transfer MQTT session streams: %v", err)
 		return

vendor/github.com/nats-io/nats-server/v2/server/opts.go

@@ -616,6 +616,9 @@ type MQTTOpts struct {
 	// PubRels).
 	AckWait time.Duration
 
+	// JSAPITimeout defines timeout for JetStream api calls (default is 5 seconds)
+	JSAPITimeout time.Duration
+
 	// MaxAckPending is the amount of QoS 1 and 2 messages (combined) the server
 	// can send to a subscription without receiving any PUBACK for those
 	// messages. The valid range is [0..65535].
@@ -5205,6 +5208,8 @@ func parseMQTT(v any, o *Options, errors *[]error, warnings *[]error) error {
 			o.MQTT.NoAuthUser = mv.(string)
 		case "ack_wait", "ackwait":
 			o.MQTT.AckWait = parseDuration("ack_wait", tk, mv, errors, warnings)
+		case "js_api_timeout", "api_timeout":
+			o.MQTT.JSAPITimeout = parseDuration("js_api_timeout", tk, mv, errors, warnings)
 		case "max_ack_pending", "max_pending", "max_inflight":
 			tmp := int(mv.(int64))
 			if tmp < 0 || tmp > 0xFFFF {

vendor/github.com/nats-io/nats-server/v2/server/raft.go

@@ -182,10 +182,11 @@ type raft struct {
 	c  *client    // Internal client for subscriptions
 	js *jetStream // JetStream, if running, to see if we are out of resources
 
-	dflag     bool        // Debug flag
-	hasleader atomic.Bool // Is there a group leader right now?
-	pleader   atomic.Bool // Has the group ever had a leader?
-	isSysAcc  atomic.Bool // Are we utilizing the system account?
+	dflag       bool        // Debug flag
+	hasleader   atomic.Bool // Is there a group leader right now?
+	pleader     atomic.Bool // Has the group ever had a leader?
+	isSysAcc    atomic.Bool // Are we utilizing the system account?
+	maybeLeader bool        // The group had a preferred leader. And is maybe already acting as leader prior to scale up.
 
 	observer bool // The node is observing, i.e. not participating in voting
 
@@ -1652,6 +1653,7 @@ func (n *raft) Campaign() error {
 func (n *raft) CampaignImmediately() error {
 	n.Lock()
 	defer n.Unlock()
+	n.maybeLeader = true
 	return n.campaign(minCampaignTimeout / 2)
 }
 
@@ -3309,6 +3311,16 @@ func (n *raft) updateLeader(newLeader string) {
 	n.hasleader.Store(newLeader != _EMPTY_)
 	if !n.pleader.Load() && newLeader != noLeader {
 		n.pleader.Store(true)
+		// If we were preferred to become the first leader, but didn't end up successful.
+		// Ensure to call lead change. When scaling from R1 to R3 we've optimized for a scale up
+		// not flipping leader/non-leader/leader status if the leader remains the same. But we need to
+		// correct that if the first leader turns out to be different.
+		if n.maybeLeader {
+			n.maybeLeader = false
+			if n.id != newLeader {
+				n.updateLeadChange(false)
+			}
+		}
 	}
 }
 

vendor/github.com/nats-io/nats-server/v2/server/stree/parts.go

@@ -123,7 +123,10 @@ func matchParts(parts [][]byte, frag []byte) ([][]byte, bool) {
 		// but update the part to what was consumed. This allows upper layers to continue.
 		if end < si+lp {
 			if end >= lf {
-				parts = append([][]byte{}, parts...) // Create a copy before modifying.
+				// Create a copy before modifying. Reuse slice capacity available at the
+				// end of the parts slice, since this saves us additional allocations.
+				lp := len(parts)
+				parts = append(parts[lp:], parts[:lp]...)
 				parts[i] = parts[i][lf-si:]
 			} else {
 				i++

vendor/github.com/nats-io/nats-server/v2/server/sublist.go

@@ -1747,23 +1747,6 @@ func IntersectStree[T any](st *stree.SubjectTree[T], sl *Sublist, cb func(subj [
 }
 
 func intersectStree[T any](st *stree.SubjectTree[T], r *level, subj []byte, cb func(subj []byte, entry *T)) {
-	// This level could potentially match literals, despite being followed up by
-	// additional wildcards. For literals we can use Find since it is considerably
-	// faster. Then we can carry on checking for further matches in the usual way.
-	wc := subjectHasWildcard(bytesToString(subj))
-	if !wc {
-		if e, ok := st.Find(subj); ok {
-			cb(subj, e)
-		}
-	}
-	if r.numNodes() == 0 {
-		// No further recursions to be made at this point but there's still a wildcard
-		// to match, so let the subject tree work it out.
-		if wc {
-			st.Match(subj, cb)
-		}
-		return
-	}
 	nsubj := subj
 	if len(nsubj) > 0 {
 		nsubj = append(subj, '.')
@@ -1779,15 +1762,28 @@ func intersectStree[T any](st *stree.SubjectTree[T], r *level, subj []byte, cb f
 		// check whether there's interest at this level (without triggering dupes) and
 		// match if so.
 		nsubj := append(nsubj, '*')
-		if len(r.pwc.psubs)+len(r.pwc.qsubs) > 0 && r.pwc.next != nil && r.pwc.next.numNodes() > 0 {
+		if len(r.pwc.psubs)+len(r.pwc.qsubs) > 0 {
 			st.Match(nsubj, cb)
 		}
-		intersectStree(st, r.pwc.next, nsubj, cb)
-	case r.numNodes() > 0:
+		if r.pwc.next != nil && r.pwc.next.numNodes() > 0 {
+			intersectStree(st, r.pwc.next, nsubj, cb)
+		}
+	default:
 		// Normal node with subject literals, keep iterating.
 		for t, n := range r.nodes {
 			nsubj := append(nsubj, t...)
-			intersectStree(st, n.next, nsubj, cb)
+			if len(n.psubs)+len(n.qsubs) > 0 {
+				if subjectHasWildcard(bytesToString(nsubj)) {
+					st.Match(nsubj, cb)
+				} else {
+					if e, ok := st.Find(nsubj); ok {
+						cb(nsubj, e)
+					}
+				}
+			}
+			if n.next != nil && n.next.numNodes() > 0 {
+				intersectStree(st, n.next, nsubj, cb)
+			}
 		}
 	}
 }