Bump github.com/nats-io/nats-server/v2 from 2.9.17 to 2.9.19

Bumps [github.com/nats-io/nats-server/v2](https://github.com/nats-io/nats-server) from 2.9.17 to 2.9.19.
- [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.9.17...v2.9.19)

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

Signed-off-by: dependabot[bot] <support@github.com>

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

@@ -958,7 +958,7 @@ func (a *Account) isLeafNodeClusterIsolated(cluster string) bool {
 	if len(a.leafClusters) > 1 {
 		return false
 	}
-	return a.leafClusters[cluster] > 0
+	return a.leafClusters[cluster] == uint64(a.nleafs)
 }
 
 // Helper function to remove leaf nodes. If number of leafnodes gets large
@@ -1429,7 +1429,7 @@ func (a *Account) sendTrackingLatency(si *serviceImport, responder *client) bool
 	}
 	sl.RequestStart = time.Unix(0, si.ts-int64(reqRTT)).UTC()
 	sl.ServiceLatency = serviceRTT - respRTT
-	sl.TotalLatency = sl.Requestor.RTT + serviceRTT
+	sl.TotalLatency = reqRTT + serviceRTT
 	if respRTT > 0 {
 		sl.SystemLatency = time.Since(ts)
 		sl.TotalLatency += sl.SystemLatency
@@ -3784,10 +3784,11 @@ func (ur *URLAccResolver) Fetch(name string) (string, error) {
 		return _EMPTY_, fmt.Errorf("could not fetch <%q>: %v", redactURLString(url), err)
 	} else if resp == nil {
 		return _EMPTY_, fmt.Errorf("could not fetch <%q>: no response", redactURLString(url))
-	} else if resp.StatusCode != http.StatusOK {
-		return _EMPTY_, fmt.Errorf("could not fetch <%q>: %v", redactURLString(url), resp.Status)
 	}
 	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return _EMPTY_, fmt.Errorf("could not fetch <%q>: %v", redactURLString(url), resp.Status)
+	}
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return _EMPTY_, err

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

@@ -1989,7 +1989,9 @@ func (c *client) authViolation() {
 			ErrAuthentication.Error(),
 			c.opts.Username)
 	} else {
-		c.Errorf(ErrAuthentication.Error())
+		if c.srv != nil {
+			c.Errorf(ErrAuthentication.Error())
+		}
 	}
 	if c.isMqtt() {
 		c.mqttEnqueueConnAck(mqttConnAckRCNotAuthorized, false)
@@ -2165,6 +2167,25 @@ func (c *client) generateClientInfoJSON(info Info) []byte {
 	info.MaxPayload = c.mpay
 	if c.isWebsocket() {
 		info.ClientConnectURLs = info.WSConnectURLs
+		if c.srv != nil { // Otherwise lame duck info can panic
+			c.srv.websocket.mu.RLock()
+			info.TLSAvailable = c.srv.websocket.tls
+			if c.srv.websocket.server != nil {
+				if tc := c.srv.websocket.server.TLSConfig; tc != nil {
+					info.TLSRequired = !tc.InsecureSkipVerify
+				}
+			}
+			if c.srv.websocket.listener != nil {
+				laddr := c.srv.websocket.listener.Addr().String()
+				if h, p, err := net.SplitHostPort(laddr); err == nil {
+					if p, err := strconv.Atoi(p); err == nil {
+						info.Host = h
+						info.Port = p
+					}
+				}
+			}
+			c.srv.websocket.mu.RUnlock()
+		}
 	}
 	info.WSConnectURLs = nil
 	// Generate the info json

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

@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.9.17"
+	VERSION = "2.9.19"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original

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

@@ -1417,9 +1417,10 @@ func (o *consumer) deleteNotActive() {
 				defer ticker.Stop()
 				for range ticker.C {
 					js.mu.RLock()
-					ca := js.consumerAssignment(acc, stream, name)
+					nca := js.consumerAssignment(acc, stream, name)
 					js.mu.RUnlock()
-					if ca != nil {
+					// Make sure this is not a new consumer with the same name.
+					if nca != nil && nca == ca {
 						s.Warnf("Consumer assignment for '%s > %s > %s' not cleaned up, retrying", acc, stream, name)
 						meta.ForwardProposal(removeEntry)
 					} else {
@@ -3255,10 +3256,10 @@ func (o *consumer) hbTimer() (time.Duration, *time.Timer) {
 // Should only be called from consumer leader.
 func (o *consumer) checkAckFloor() {
 	o.mu.RLock()
-	mset, closed, asflr := o.mset, o.closed, o.asflr
+	mset, closed, asflr, numPending := o.mset, o.closed, o.asflr, len(o.pending)
 	o.mu.RUnlock()
 
-	if closed || mset == nil {
+	if asflr == 0 || closed || mset == nil {
 		return
 	}
 
@@ -3270,19 +3271,46 @@ func (o *consumer) checkAckFloor() {
 		return
 	}
 
-	// Process all messages that no longer exist.
-	for seq := asflr + 1; seq < ss.FirstSeq; seq++ {
-		// Check if this message was pending.
+	// Check which linear space is less to walk.
+	if ss.FirstSeq-asflr-1 < uint64(numPending) {
+		// Process all messages that no longer exist.
+		for seq := asflr + 1; seq < ss.FirstSeq; seq++ {
+			// Check if this message was pending.
+			o.mu.RLock()
+			p, isPending := o.pending[seq]
+			var rdc uint64 = 1
+			if o.rdc != nil {
+				rdc = o.rdc[seq]
+			}
+			o.mu.RUnlock()
+			// If it was pending for us, get rid of it.
+			if isPending {
+				o.processTerm(seq, p.Sequence, rdc)
+			}
+		}
+	} else if numPending > 0 {
+		// here it shorter to walk pending.
+		// toTerm is seq, dseq, rcd for each entry.
+		toTerm := make([]uint64, 0, numPending*3)
 		o.mu.RLock()
-		p, isPending := o.pending[seq]
-		var rdc uint64 = 1
-		if o.rdc != nil {
-			rdc = o.rdc[seq]
+		for seq, p := range o.pending {
+			if seq < ss.FirstSeq {
+				var dseq uint64 = 1
+				if p != nil {
+					dseq = p.Sequence
+				}
+				var rdc uint64 = 1
+				if o.rdc != nil {
+					rdc = o.rdc[seq]
+				}
+				toTerm = append(toTerm, seq, dseq, rdc)
+			}
 		}
 		o.mu.RUnlock()
-		// If it was pending for us, get rid of it.
-		if isPending {
-			o.processTerm(seq, p.Sequence, rdc)
+
+		for i := 0; i < len(toTerm); i += 3 {
+			seq, dseq, rdc := toTerm[i], toTerm[i+1], toTerm[i+2]
+			o.processTerm(seq, dseq, rdc)
 		}
 	}
 
@@ -3315,15 +3343,21 @@ func (o *consumer) checkAckFloor() {
 func (o *consumer) processInboundAcks(qch chan struct{}) {
 	// Grab the server lock to watch for server quit.
 	o.mu.RLock()
-	s := o.srv
+	s, mset := o.srv, o.mset
 	hasInactiveThresh := o.cfg.InactiveThreshold > 0
 	o.mu.RUnlock()
 
+	if s == nil || mset == nil {
+		return
+	}
+
 	// We will check this on entry and periodically.
 	o.checkAckFloor()
 
 	// How often we will check for ack floor drift.
-	var ackFloorCheck = 30 * time.Second
+	// Spread these out for large numbers on a server restart.
+	delta := time.Duration(rand.Int63n(int64(time.Minute)))
+	var ackFloorCheck = time.Minute + delta
 
 	for {
 		select {

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

@@ -99,25 +99,26 @@ type inSysMsg struct {
 
 // Used to send and receive messages from inside the server.
 type internal struct {
-	account  *Account
-	client   *client
-	seq      uint64
-	sid      int
-	servers  map[string]*serverUpdate
-	sweeper  *time.Timer
-	stmr     *time.Timer
-	replies  map[string]msgHandler
-	sendq    *ipQueue[*pubMsg]
-	recvq    *ipQueue[*inSysMsg]
-	resetCh  chan struct{}
-	wg       sync.WaitGroup
-	sq       *sendq
-	orphMax  time.Duration
-	chkOrph  time.Duration
-	statsz   time.Duration
-	cstatsz  time.Duration
-	shash    string
-	inboxPre string
+	account        *Account
+	client         *client
+	seq            uint64
+	sid            int
+	servers        map[string]*serverUpdate
+	sweeper        *time.Timer
+	stmr           *time.Timer
+	replies        map[string]msgHandler
+	sendq          *ipQueue[*pubMsg]
+	recvq          *ipQueue[*inSysMsg]
+	resetCh        chan struct{}
+	wg             sync.WaitGroup
+	sq             *sendq
+	orphMax        time.Duration
+	chkOrph        time.Duration
+	statsz         time.Duration
+	cstatsz        time.Duration
+	shash          string
+	inboxPre       string
+	remoteStatsSub *subscription
 }
 
 // ServerStatsMsg is sent periodically with stats updates.
@@ -642,8 +643,6 @@ func (s *Server) checkRemoteServers() {
 // Grab RSS and PCPU
 // Server lock will be held but released.
 func (s *Server) updateServerUsage(v *ServerStats) {
-	s.mu.Unlock()
-	defer s.mu.Lock()
 	var vss int64
 	pse.ProcUsage(&v.CPU, &v.Mem, &vss)
 	v.Cores = runtime.NumCPU()
@@ -679,6 +678,32 @@ func routeStat(r *client) *RouteStat {
 func (s *Server) sendStatsz(subj string) {
 	var m ServerStatsMsg
 	s.updateServerUsage(&m.Stats)
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	// Check that we have a system account, etc.
+	if s.sys == nil || s.sys.account == nil {
+		return
+	}
+
+	// if we are running standalone, check for interest.
+	if s.standAloneMode() {
+		// Check if we even have interest in this subject.
+		sacc := s.sys.account
+		rr := sacc.sl.Match(subj)
+		totalSubs := len(rr.psubs) + len(rr.qsubs)
+		if totalSubs == 0 {
+			return
+		} else if totalSubs == 1 && len(rr.psubs) == 1 {
+			// For the broadcast subject we listen to that ourselves with no echo for remote updates.
+			// If we are the only ones listening do not send either.
+			if rr.psubs[0] == s.sys.remoteStatsSub {
+				return
+			}
+		}
+	}
+
 	m.Stats.Start = s.start
 	m.Stats.Connections = len(s.clients)
 	m.Stats.TotalConnections = s.totalClients
@@ -722,14 +747,12 @@ func (s *Server) sendStatsz(subj string) {
 		gw.RUnlock()
 	}
 	// Active Servers
-	m.Stats.ActiveServers = 1
-	if s.sys != nil {
-		m.Stats.ActiveServers += len(s.sys.servers)
-	}
+	m.Stats.ActiveServers = len(s.sys.servers) + 1
+
 	// JetStream
 	if js := s.js; js != nil {
 		jStat := &JetStreamVarz{}
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		js.mu.RLock()
 		c := js.config
 		c.StoreDir = _EMPTY_
@@ -771,7 +794,7 @@ func (s *Server) sendStatsz(subj string) {
 			}
 		}
 		m.Stats.JetStream = jStat
-		s.mu.Lock()
+		s.mu.RLock()
 	}
 	// Send message.
 	s.sendInternalMsg(subj, _EMPTY_, &m.Server, &m)
@@ -790,13 +813,12 @@ func (s *Server) heartbeatStatsz() {
 		}
 		s.sys.stmr.Reset(s.sys.cstatsz)
 	}
-	s.sendStatsz(fmt.Sprintf(serverStatsSubj, s.info.ID))
+	// Do in separate Go routine.
+	go s.sendStatszUpdate()
 }
 
 func (s *Server) sendStatszUpdate() {
-	s.mu.Lock()
-	s.sendStatsz(fmt.Sprintf(serverStatsSubj, s.info.ID))
-	s.mu.Unlock()
+	s.sendStatsz(fmt.Sprintf(serverStatsSubj, s.ID()))
 }
 
 // This should be wrapChk() to setup common locking.
@@ -897,8 +919,11 @@ func (s *Server) initEventTracking() {
 	}
 	// Listen for statsz from others.
 	subject = fmt.Sprintf(serverStatsSubj, "*")
-	if _, err := s.sysSubscribe(subject, s.noInlineCallback(s.remoteServerUpdate)); err != nil {
+	if sub, err := s.sysSubscribe(subject, s.noInlineCallback(s.remoteServerUpdate)); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
+	} else {
+		// Keep track of this one.
+		s.sys.remoteStatsSub = sub
 	}
 	// Listen for all server shutdowns.
 	subject = fmt.Sprintf(shutdownEventSubj, "*")
@@ -1338,7 +1363,8 @@ func (s *Server) processNewServer(si *ServerInfo) {
 		}
 	}
 	// Announce ourselves..
-	s.sendStatsz(fmt.Sprintf(serverStatsSubj, s.info.ID))
+	// Do this in a separate Go routine.
+	go s.sendStatszUpdate()
 }
 
 // If GW is enabled on this server and there are any leaf node connections,
@@ -1612,9 +1638,7 @@ func (s *Server) statszReq(sub *subscription, c *client, _ *Account, subject, re
 			return
 		}
 	}
-	s.mu.Lock()
 	s.sendStatsz(reply)
-	s.mu.Unlock()
 }
 
 var errSkipZreq = errors.New("filtered response")

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

@@ -4665,7 +4665,16 @@ func (fs *fileStore) loadLast(subj string, sm *StoreMsg) (lsm *StoreMsg, err err
 			mb.mu.Unlock()
 			return nil, err
 		}
-		_, _, l := mb.filteredPendingLocked(subj, wc, mb.first.seq)
+		var l uint64
+		// Optimize if subject is not a wildcard.
+		if !wc {
+			if ss := mb.fss[subj]; ss != nil {
+				l = ss.Last
+			}
+		}
+		if l == 0 {
+			_, _, l = mb.filteredPendingLocked(subj, wc, mb.first.seq)
+		}
 		if l > 0 {
 			if mb.cacheNotLoaded() {
 				if err := mb.loadMsgsWithLock(); err != nil {
@@ -5108,10 +5117,6 @@ func compareFn(subject string) func(string, string) bool {
 // PurgeEx will remove messages based on subject filters, sequence and number of messages to keep.
 // Will return the number of purged messages.
 func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
-	if sequence > 1 && keep > 0 {
-		return 0, ErrPurgeArgMismatch
-	}
-
 	if subject == _EMPTY_ || subject == fwcs {
 		if keep == 0 && (sequence == 0 || sequence == 1) {
 			return fs.Purge()

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

@@ -870,9 +870,7 @@ func (s *Server) createGateway(cfg *gatewayCfg, url *url.URL, conn net.Conn) {
 
 	// Announce ourselves again to new connections.
 	if solicit && s.EventsEnabled() {
-		s.mu.Lock()
-		s.sendStatsz(fmt.Sprintf(serverStatsSubj, s.info.ID))
-		s.mu.Unlock()
+		s.sendStatszUpdate()
 	}
 }
 

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

@@ -104,6 +104,9 @@ func (q *ipQueue[T]) push(e T) int {
 // emptied the queue. So the caller should never assume that pop() will
 // return a slice of 1 or more, it could return `nil`.
 func (q *ipQueue[T]) pop() []T {
+	if q == nil {
+		return nil
+	}
 	var elts []T
 	q.Lock()
 	if q.pos == 0 {

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

@@ -1020,9 +1020,9 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 		return fmt.Errorf("jetstream can not be enabled on the system account")
 	}
 
-	s.mu.Lock()
+	s.mu.RLock()
 	sendq := s.sys.sendq
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	// No limits means we dynamically set up limits.
 	// We also place limits here so we know that the account is configured for JetStream.
@@ -1054,12 +1054,15 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 	jsa := &jsAccount{js: js, account: a, limits: limits, streams: make(map[string]*stream), sendq: sendq, usage: make(map[string]*jsaStorage)}
 	jsa.storeDir = filepath.Join(js.config.StoreDir, a.Name)
 
-	jsa.usageMu.Lock()
-	jsa.utimer = time.AfterFunc(usageTick, jsa.sendClusterUsageUpdateTimer)
-	// Cluster mode updates to resource usage, but we always will turn on. System internal prevents echos.
-	jsa.updatesPub = fmt.Sprintf(jsaUpdatesPubT, a.Name, sysNode)
-	jsa.updatesSub, _ = s.sysSubscribe(fmt.Sprintf(jsaUpdatesSubT, a.Name), jsa.remoteUpdateUsage)
-	jsa.usageMu.Unlock()
+	// A single server does not need to do the account updates at this point.
+	if js.cluster != nil || !s.standAloneMode() {
+		jsa.usageMu.Lock()
+		jsa.utimer = time.AfterFunc(usageTick, jsa.sendClusterUsageUpdateTimer)
+		// Cluster mode updates to resource usage. System internal prevents echos.
+		jsa.updatesPub = fmt.Sprintf(jsaUpdatesPubT, a.Name, sysNode)
+		jsa.updatesSub, _ = s.sysSubscribe(fmt.Sprintf(jsaUpdatesSubT, a.Name), jsa.remoteUpdateUsage)
+		jsa.usageMu.Unlock()
+	}
 
 	js.accounts[a.Name] = jsa
 	js.mu.Unlock()
@@ -1209,22 +1212,23 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 
 		// Check if we are encrypted.
 		keyFile := filepath.Join(mdir, JetStreamMetaFileKey)
-		if key, err := os.ReadFile(keyFile); err == nil {
+		keyBuf, err := os.ReadFile(keyFile)
+		if err == nil {
 			s.Debugf("  Stream metafile is encrypted, reading encrypted keyfile")
-			if len(key) < minMetaKeySize {
-				s.Warnf("  Bad stream encryption key length of %d", len(key))
+			if len(keyBuf) < minMetaKeySize {
+				s.Warnf("  Bad stream encryption key length of %d", len(keyBuf))
 				continue
 			}
 			// Decode the buffer before proceeding.
-			nbuf, err := s.decryptMeta(sc, key, buf, a.Name, fi.Name())
+			nbuf, err := s.decryptMeta(sc, keyBuf, buf, a.Name, fi.Name())
 			if err != nil {
 				// See if we are changing ciphers.
 				switch sc {
 				case ChaCha:
-					nbuf, err = s.decryptMeta(AES, key, buf, a.Name, fi.Name())
+					nbuf, err = s.decryptMeta(AES, keyBuf, buf, a.Name, fi.Name())
 					osc, convertingCiphers = AES, true
 				case AES:
-					nbuf, err = s.decryptMeta(ChaCha, key, buf, a.Name, fi.Name())
+					nbuf, err = s.decryptMeta(ChaCha, keyBuf, buf, a.Name, fi.Name())
 					osc, convertingCiphers = ChaCha, true
 				}
 				if err != nil {
@@ -1234,9 +1238,6 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 			}
 			buf = nbuf
 			plaintext = false
-
-			// Remove the key file to have system regenerate with the new cipher.
-			os.Remove(keyFile)
 		}
 
 		var cfg FileStreamInfo
@@ -1288,6 +1289,8 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 				s.Noticef("  Encrypting stream '%s > %s'", a.Name, cfg.StreamConfig.Name)
 			} else if convertingCiphers {
 				s.Noticef("  Converting from %s to %s for stream '%s > %s'", osc, sc, a.Name, cfg.StreamConfig.Name)
+				// Remove the key file to have system regenerate with the new cipher.
+				os.Remove(keyFile)
 			}
 		}
 
@@ -1295,6 +1298,13 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 		mset, err := a.addStream(&cfg.StreamConfig)
 		if err != nil {
 			s.Warnf("  Error recreating stream %q: %v", cfg.Name, err)
+			// If we removed a keyfile from above make sure to put it back.
+			if convertingCiphers {
+				err := os.WriteFile(keyFile, keyBuf, defaultFilePerms)
+				if err != nil {
+					s.Warnf("  Error replacing meta keyfile for stream %q: %v", cfg.Name, err)
+				}
+			}
 			continue
 		}
 		if !cfg.Created.IsZero() {

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

@@ -2808,12 +2808,12 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 					panic(err.Error())
 				}
 				// Ignore if we are recovering and we have already processed.
-				if isRecovering {
-					if mset.state().FirstSeq <= sp.LastSeq {
-						// Make sure all messages from the purge are gone.
-						mset.store.Compact(sp.LastSeq + 1)
+				if isRecovering && (sp.Request == nil || sp.Request.Sequence == 0) {
+					if sp.Request == nil {
+						sp.Request = &JSApiStreamPurgeRequest{Sequence: sp.LastSeq}
+					} else {
+						sp.Request.Sequence = sp.LastSeq
 					}
-					continue
 				}
 
 				s := js.server()
@@ -3107,7 +3107,9 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) bool {
 		accStreams = make(map[string]*streamAssignment)
 	} else if osa := accStreams[stream]; osa != nil && osa != sa {
 		// Copy over private existing state from former SA.
-		sa.Group.node = osa.Group.node
+		if sa.Group != nil {
+			sa.Group.node = osa.Group.node
+		}
 		sa.consumers = osa.consumers
 		sa.responded = osa.responded
 		sa.err = osa.err
@@ -3198,7 +3200,9 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 	}
 
 	// Copy over private existing state from former SA.
-	sa.Group.node = osa.Group.node
+	if sa.Group != nil {
+		sa.Group.node = osa.Group.node
+	}
 	sa.consumers = osa.consumers
 	sa.err = osa.err
 
@@ -3216,7 +3220,9 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 		sa.responded = false
 	} else {
 		// Make sure to clean up any old node in case this stream moves back here.
-		sa.Group.node = nil
+		if sa.Group != nil {
+			sa.Group.node = nil
+		}
 	}
 	js.mu.Unlock()
 
@@ -3400,6 +3406,7 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 	s, rg := js.srv, sa.Group
 	alreadyRunning := rg.node != nil
 	storage := sa.Config.Storage
+	restore := sa.Restore
 	js.mu.RUnlock()
 
 	// Process the raft group and make sure it's running if needed.
@@ -3408,11 +3415,13 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 	// If we are restoring, create the stream if we are R>1 and not the preferred who handles the
 	// receipt of the snapshot itself.
 	shouldCreate := true
-	if sa.Restore != nil {
+	if restore != nil {
 		if len(rg.Peers) == 1 || rg.node != nil && rg.node.ID() == rg.Preferred {
 			shouldCreate = false
 		} else {
+			js.mu.Lock()
 			sa.Restore = nil
+			js.mu.Unlock()
 		}
 	}
 
@@ -3784,7 +3793,9 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 	} else if oca := sa.consumers[ca.Name]; oca != nil {
 		wasExisting = true
 		// Copy over private existing state from former SA.
-		ca.Group.node = oca.Group.node
+		if ca.Group != nil {
+			ca.Group.node = oca.Group.node
+		}
 		ca.responded = oca.responded
 		ca.err = oca.err
 	}
@@ -3899,8 +3910,12 @@ func (js *jetStream) processConsumerRemoval(ca *consumerAssignment) {
 	var needDelete bool
 	if accStreams := cc.streams[ca.Client.serviceAccount()]; accStreams != nil {
 		if sa := accStreams[ca.Stream]; sa != nil && sa.consumers != nil && sa.consumers[ca.Name] != nil {
-			needDelete = true
-			delete(sa.consumers, ca.Name)
+			oca := sa.consumers[ca.Name]
+			// Make sure this removal is for what we have, otherwise ignore.
+			if ca.Group != nil && oca.Group != nil && ca.Group.Name == oca.Group.Name {
+				needDelete = true
+				delete(sa.consumers, ca.Name)
+			}
 		}
 	}
 	js.mu.Unlock()

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

@@ -1422,7 +1422,7 @@ func (s *Server) addLeafNodeConnection(c *client, srvName, clusterName string, c
 	}
 	// If we have a specified JetStream domain we will want to add a mapping to
 	// allow access cross domain for each non-system account.
-	if opts.JetStreamDomain != _EMPTY_ && acc != sysAcc && opts.JetStream {
+	if opts.JetStreamDomain != _EMPTY_ && opts.JetStream && acc != nil && acc != sysAcc {
 		for src, dest := range generateJSMappingTable(opts.JetStreamDomain) {
 			if err := acc.AddMapping(src, dest); err != nil {
 				c.Debugf("Error adding JetStream domain mapping: %s", err.Error())

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

@@ -614,10 +614,6 @@ func (ms *memStore) expireMsgs() {
 // PurgeEx will remove messages based on subject filters, sequence and number of messages to keep.
 // Will return the number of purged messages.
 func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
-	if sequence > 1 && keep > 0 {
-		return 0, ErrPurgeArgMismatch
-	}
-
 	if subject == _EMPTY_ || subject == fwcs {
 		if keep == 0 && (sequence == 0 || sequence == 1) {
 			return ms.Purge()
@@ -868,6 +864,10 @@ func (ms *memStore) LoadLastMsg(subject string, smp *StoreMsg) (*StoreMsg, error
 
 	if subject == _EMPTY_ || subject == fwcs {
 		sm, ok = ms.msgs[ms.state.LastSeq]
+	} else if subjectIsLiteral(subject) {
+		if ss := ms.fss[subject]; ss != nil && ss.Msgs > 0 {
+			sm, ok = ms.msgs[ss.Last]
+		}
 	} else if ss := ms.filteredStateLocked(1, subject, true); ss.Msgs > 0 {
 		sm, ok = ms.msgs[ss.Last]
 	}
@@ -899,8 +899,8 @@ func (ms *memStore) LoadNextMsg(filter string, wc bool, start uint64, smp *Store
 
 	isAll := filter == _EMPTY_ || filter == fwcs
 
-	// Skip scan of mb.fss is number of messages in the block are less than
-	// 1/2 the number of subjects in mb.fss. Or we have a wc and lots of fss entries.
+	// Skip scan of ms.fss is number of messages in the block are less than
+	// 1/2 the number of subjects in ms.fss. Or we have a wc and lots of fss entries.
 	const linearScanMaxFSS = 256
 	doLinearScan := isAll || 2*int(ms.state.LastSeq-start) < len(ms.fss) || (wc && len(ms.fss) > linearScanMaxFSS)
 

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

@@ -1474,12 +1474,6 @@ func (n *raft) Peers() []*Peer {
 // Update our known set of peers.
 func (n *raft) UpdateKnownPeers(knownPeers []string) {
 	n.Lock()
-	// If this is a scale up, let the normal add peer logic take precedence.
-	// Otherwise if the new peers are slow to start we stall ourselves.
-	if len(knownPeers) > len(n.peers) {
-		n.Unlock()
-		return
-	}
 	// Process like peer state update.
 	ps := &peerState{knownPeers, len(knownPeers), n.extSt}
 	n.processPeerState(ps)
@@ -1816,9 +1810,11 @@ func (n *raft) runAsFollower() {
 			} else if n.isCatchingUp() {
 				n.debug("Not switching to candidate, catching up")
 				// Check to see if our catchup has stalled.
+				n.Lock()
 				if n.catchupStalled() {
 					n.cancelCatchup()
 				}
+				n.Unlock()
 			} else {
 				n.switchToCandidate()
 				return

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

@@ -1949,12 +1949,12 @@ func (c *client) processRouteConnect(srv *Server, arg []byte, lang string) error
 		c.closeConnection(WrongGateway)
 		return ErrWrongGateway
 	}
-	var perms *RoutePermissions
-	//TODO this check indicates srv may be nil. see srv usage below
-	if srv != nil {
-		perms = srv.getOpts().Cluster.Permissions
+
+	if srv == nil {
+		return ErrServerNotRunning
 	}
 
+	perms := srv.getOpts().Cluster.Permissions
 	clusterName := srv.ClusterName()
 
 	// If we have a cluster name set, make sure it matches ours.

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

@@ -3560,7 +3560,10 @@ func (s *Server) lameDuckMode() {
 	numClients := int64(len(s.clients))
 	batch := 1
 	// Sleep interval between each client connection close.
-	si := dur / numClients
+	var si int64
+	if numClients != 0 {
+		si = dur / numClients
+	}
 	if si < 1 {
 		// Should not happen (except in test with very small LD duration), but
 		// if there are too many clients, batch the number of close and

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

@@ -61,8 +61,6 @@ var (
 	ErrInvalidSequence = errors.New("invalid sequence")
 	// ErrSequenceMismatch is returned when storing a raw message and the expected sequence is wrong.
 	ErrSequenceMismatch = errors.New("expected sequence does not match store")
-	// ErrPurgeArgMismatch is returned when PurgeEx is called with sequence > 1 and keep > 0.
-	ErrPurgeArgMismatch = errors.New("sequence > 1 && keep > 0 not allowed")
 )
 
 // StoreMsg is the stored message format for messages that are retained by the Store layer.

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

@@ -1671,7 +1671,7 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
 		mset.mu.RUnlock()
 		return 0, errors.New("sealed stream")
 	}
-	store := mset.store
+	store, mlseq := mset.store, mset.lseq
 	mset.mu.RUnlock()
 
 	if preq != nil {
@@ -1683,11 +1683,17 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
 		return purged, err
 	}
 
-	// Purge consumers.
+	// Grab our stream state.
 	var state StreamState
 	store.FastState(&state)
 	fseq, lseq := state.FirstSeq, state.LastSeq
 
+	// Check if our last has moved past what our original last sequence was, if so reset.
+	if lseq > mlseq {
+		mset.setLastSeq(lseq)
+	}
+
+	// Purge consumers.
 	// Check for filtered purge.
 	if preq != nil && preq.Subject != _EMPTY_ {
 		ss := store.FilteredState(state.FirstSeq, preq.Subject)
@@ -1788,7 +1794,9 @@ func gatherSourceMirrorSubjects(subjects []string, cfg *StreamConfig, acc *Accou
 
 // Return the subjects for a stream source.
 func (a *Account) streamSourceSubjects(ss *StreamSource, seen map[string]bool) (subjects []string, hasExt bool) {
-	if ss != nil && ss.External != nil {
+	if ss == nil {
+		return nil, false
+	} else if ss.External != nil {
 		return nil, true
 	}
 
@@ -2397,7 +2405,14 @@ func (mset *stream) setupMirrorConsumer() error {
 
 				// Check if we need to skip messages.
 				if state.LastSeq != ccr.ConsumerInfo.Delivered.Stream {
-					mset.skipMsgs(state.LastSeq+1, ccr.ConsumerInfo.Delivered.Stream)
+					// Check to see if delivered is past our last and we have no msgs. This will help the
+					// case when mirroring a stream that has a very high starting sequence number.
+					if state.Msgs == 0 && ccr.ConsumerInfo.Delivered.Stream > state.LastSeq {
+						mset.store.PurgeEx(_EMPTY_, ccr.ConsumerInfo.Delivered.Stream+1, 0)
+						mset.lseq = ccr.ConsumerInfo.Delivered.Stream
+					} else {
+						mset.skipMsgs(state.LastSeq+1, ccr.ConsumerInfo.Delivered.Stream)
+					}
 				}
 
 				// Capture consumer name.
@@ -4490,7 +4505,9 @@ func (mset *stream) resetAndWaitOnConsumers() {
 			}
 			node.Delete()
 		}
-		o.monitorWg.Wait()
+		if o.isMonitorRunning() {
+			o.monitorWg.Wait()
+		}
 	}
 }
 

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

@@ -615,7 +615,7 @@ func (s *Sublist) reduceCacheCount() {
 
 // Helper function for auto-expanding remote qsubs.
 func isRemoteQSub(sub *subscription) bool {
-	return sub != nil && sub.queue != nil && sub.client != nil && sub.client.kind == ROUTER
+	return sub != nil && sub.queue != nil && sub.client != nil && (sub.client.kind == ROUTER || sub.client.kind == LEAF)
 }
 
 // UpdateRemoteQSub should be called when we update the weight of an existing

vendor/github.com/nats-io/nats.go/.golangci.yaml

@@ -0,0 +1,10 @@
+issues:
+  max-issues-per-linter: 0
+  max-same-issues: 0
+  exclude-rules:
+    - linters:
+      - errcheck
+      text: "Unsubscribe"
+    - linters:
+      - errcheck
+      text: "msg.Ack"

vendor/github.com/nats-io/nats.go/.travis.yml

@@ -5,6 +5,7 @@ go:
 go_import_path: github.com/nats-io/nats.go
 install:
 - go get -t ./...
+- curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin
 - if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then
     go install github.com/mattn/goveralls@latest;
     go install github.com/wadey/gocovmerge@latest;
@@ -18,6 +19,7 @@ before_script:
     find . -type f -name "*.go" | xargs misspell -error -locale US;
     GOFLAGS="-mod=mod -modfile=go_test.mod" staticcheck ./...;
   fi
+- golangci-lint run ./jetstream/...
 script:
 - go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
 - if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi

vendor/github.com/nats-io/nats.go/README.md

@@ -29,7 +29,7 @@ When using or transitioning to Go modules support:
 ```bash
 # Go client latest or explicit version
 go get github.com/nats-io/nats.go/@latest
-go get github.com/nats-io/nats.go/@v1.24.0
+go get github.com/nats-io/nats.go/@v1.27.0
 
 # For latest NATS Server, add /v2 at the end
 go get github.com/nats-io/nats-server/v2
@@ -92,6 +92,13 @@ nc.Close()
 
 ## JetStream Basic Usage
 
+> __NOTE__
+>
+> We encourage you to try out a new, simplified version on JetStream API.
+> The new API is currently in preview and is available under `jetstream` package.
+>
+> You can find more information on the new API [here](https://github.com/nats-io/nats.go/blob/main/jetstream/README.md)
+
 ```go
 import "github.com/nats-io/nats.go"
 

vendor/github.com/nats-io/nats.go/context.go

@@ -136,9 +136,8 @@ func (s *Subscription) nextMsgWithContext(ctx context.Context, pullSubInternal,
 		}
 		if err := s.processNextMsgDelivered(msg); err != nil {
 			return nil, err
-		} else {
-			return msg, nil
 		}
+		return msg, nil
 	default:
 		// If internal and we don't want to wait, signal that there is no
 		// message in the internal queue.

vendor/github.com/nats-io/nats.go/enc.go

@@ -265,5 +265,5 @@ func (c *EncodedConn) Drain() error {
 
 // LastError reports the last error encountered via the Connection.
 func (c *EncodedConn) LastError() error {
-	return c.Conn.err
+	return c.Conn.LastError()
 }

vendor/github.com/nats-io/nats.go/go_test.mod

@@ -4,17 +4,19 @@ go 1.19
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/nats-io/nats-server/v2 v2.9.6
-	github.com/nats-io/nkeys v0.3.0
+	github.com/klauspost/compress v1.16.5
+	github.com/nats-io/nats-server/v2 v2.9.16
+	github.com/nats-io/nkeys v0.4.4
 	github.com/nats-io/nuid v1.0.1
+	go.uber.org/goleak v1.2.1
+	golang.org/x/text v0.9.0
 	google.golang.org/protobuf v1.23.0
 )
 
 require (
-	github.com/klauspost/compress v1.15.11 // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect
-	github.com/nats-io/jwt/v2 v2.3.0 // indirect
-	golang.org/x/crypto v0.5.0 // indirect
-	golang.org/x/sys v0.4.0 // indirect
-	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af // indirect
+	github.com/nats-io/jwt/v2 v2.4.1 // indirect
+	golang.org/x/crypto v0.8.0 // indirect
+	golang.org/x/sys v0.7.0 // indirect
+	golang.org/x/time v0.3.0 // indirect
 )

vendor/github.com/nats-io/nats.go/go_test.sum

@@ -1,3 +1,4 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
 github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
 github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
@@ -9,31 +10,31 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B9Sx9c=
-github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
+github.com/klauspost/compress v1.16.5 h1:IFV2oUNUzZaz+XyusxpLzpzS8Pt5rh0Z16For/djlyI=
+github.com/klauspost/compress v1.16.5/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
-github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
-github.com/nats-io/jwt/v2 v2.3.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
-github.com/nats-io/nats-server/v2 v2.9.6 h1:RTtK+rv/4CcliOuqGsy58g7MuWkBaWmF5TUNwuUo9Uw=
-github.com/nats-io/nats-server/v2 v2.9.6/go.mod h1:AB6hAnGZDlYfqb7CTAm66ZKMZy9DpfierY1/PbpvI2g=
-github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
-github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
+github.com/nats-io/jwt/v2 v2.4.1 h1:Y35W1dgbbz2SQUYDPCaclXcuqleVmpbRa7646Jf2EX4=
+github.com/nats-io/jwt/v2 v2.4.1/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
+github.com/nats-io/nats-server/v2 v2.9.16 h1:SuNe6AyCcVy0g5326wtyU8TdqYmcPqzTjhkHojAjprc=
+github.com/nats-io/nats-server/v2 v2.9.16/go.mod h1:z1cc5Q+kqJkz9mLUdlcSsdYnId4pyImHjNgoh6zxSC0=
+github.com/nats-io/nkeys v0.4.4 h1:xvBJ8d69TznjcQl9t6//Q5xXuVhyYiSos6RPtvQNTwA=
+github.com/nats-io/nkeys v0.4.4/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/crypto v0.5.0 h1:U/0M97KRkSFvyD/3FSmdP5W5swImpNgle/EHFhOsQPE=
-golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
-golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
+go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
+golang.org/x/crypto v0.8.0 h1:pd9TJtTueMTVQXzk8E2XESSMQDj/U7OUu0PqJqPXQjQ=
+golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.4.0 h1:Zr2JFtRQNX3BCZ8YtxRE9hNJYC8J6I1MVbMg6owUp18=
-golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/time v0.0.0-20220922220347-f3bd1da661af h1:Yx9k8YCG3dvF87UAn2tu2HQLf2dt/eR1bXxpLMWeH+Y=
-golang.org/x/time v0.0.0-20220922220347-f3bd1da661af/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
-golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
+golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
+golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
@@ -43,3 +44,4 @@ google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miE
 google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
 google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

vendor/github.com/nats-io/nats.go/internal/parser/parse.go

@@ -0,0 +1,104 @@
+// Copyright 2020-2022 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package parser
+
+import (
+	"errors"
+	"fmt"
+)
+
+const (
+	AckDomainTokenPos = iota + 2
+	AckAccHashTokenPos
+	AckStreamTokenPos
+	AckConsumerTokenPos
+	AckNumDeliveredTokenPos
+	AckStreamSeqTokenPos
+	AckConsumerSeqTokenPos
+	AckTimestampSeqTokenPos
+	AckNumPendingTokenPos
+)
+
+var ErrInvalidSubjectFormat = errors.New("invalid format of ACK subject")
+
+// Quick parser for positive numbers in ack reply encoding.
+// NOTE: This parser does not detect uint64 overflow
+func ParseNum(d string) (n uint64) {
+	if len(d) == 0 {
+		return 0
+	}
+
+	// ASCII numbers 0-9
+	const (
+		asciiZero = 48
+		asciiNine = 57
+	)
+
+	for _, dec := range d {
+		if dec < asciiZero || dec > asciiNine {
+			return 0
+		}
+		n = n*10 + uint64(dec) - asciiZero
+	}
+	return
+}
+
+func GetMetadataFields(subject string) ([]string, error) {
+	v1TokenCounts, v2TokenCounts := 9, 12
+
+	var start int
+	tokens := make([]string, 0, v2TokenCounts)
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == '.' {
+			tokens = append(tokens, subject[start:i])
+			start = i + 1
+		}
+	}
+	tokens = append(tokens, subject[start:])
+	//
+	// Newer server will include the domain name and account hash in the subject,
+	// and a token at the end.
+	//
+	// Old subject was:
+	// $JS.ACK.<stream>.<consumer>.<delivered>.<sseq>.<cseq>.<tm>.<pending>
+	//
+	// New subject would be:
+	// $JS.ACK.<domain>.<account hash>.<stream>.<consumer>.<delivered>.<sseq>.<cseq>.<tm>.<pending>.<a token with a random value>
+	//
+	// v1 has 9 tokens, v2 has 12, but we must not be strict on the 12th since
+	// it may be removed in the future. Also, the library has no use for it.
+	// The point is that a v2 ACK subject is valid if it has at least 11 tokens.
+	//
+	tokensLen := len(tokens)
+	// If lower than 9 or more than 9 but less than 11, report an error
+	if tokensLen < v1TokenCounts || (tokensLen > v1TokenCounts && tokensLen < v2TokenCounts-1) {
+		return nil, ErrInvalidSubjectFormat
+	}
+	if tokens[0] != "$JS" || tokens[1] != "ACK" {
+		return nil, fmt.Errorf("%w: subject should start with $JS.ACK", ErrInvalidSubjectFormat)
+	}
+	// For v1 style, we insert 2 empty tokens (domain and hash) so that the
+	// rest of the library references known fields at a constant location.
+	if tokensLen == v1TokenCounts {
+		// Extend the array (we know the backend is big enough)
+		tokens = append(tokens[:AckDomainTokenPos+2], tokens[AckDomainTokenPos:]...)
+		// Clear the domain and hash tokens
+		tokens[AckDomainTokenPos], tokens[AckAccHashTokenPos] = "", ""
+
+	} else if tokens[AckDomainTokenPos] == "_" {
+		// If domain is "_", replace with empty value.
+		tokens[AckDomainTokenPos] = ""
+	}
+	return tokens, nil
+}

vendor/github.com/nats-io/nats.go/js.go

@@ -27,6 +27,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/nats-io/nats.go/internal/parser"
 	"github.com/nats-io/nuid"
 )
 
@@ -678,6 +679,15 @@ func (js *js) newAsyncReply() string {
 	return sb.String()
 }
 
+func (js *js) cleanupReplySub() {
+	js.mu.Lock()
+	if js.rsub != nil {
+		js.rsub.Unsubscribe()
+		js.rsub = nil
+	}
+	js.mu.Unlock()
+}
+
 // registerPAF will register for a PubAckFuture.
 func (js *js) registerPAF(id string, paf *pubAckFuture) (int, int) {
 	js.mu.Lock()
@@ -1468,6 +1478,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		isDurable     = o.cfg.Durable != _EMPTY_
 		consumerBound = o.bound
 		ctx           = o.ctx
+		skipCInfo     = o.skipCInfo
 		notFoundErr   bool
 		lookupErr     bool
 		nc            = js.nc
@@ -1541,8 +1552,8 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 
 	// With an explicit durable name, we can lookup the consumer first
 	// to which it should be attaching to.
-	// If bind to ordered consumer is true, skip the lookup.
-	if consumer != _EMPTY_ {
+	// If SkipConsumerLookup was used, do not call consumer info.
+	if consumer != _EMPTY_ && !o.skipCInfo {
 		info, err = js.ConsumerInfo(stream, consumer)
 		notFoundErr = errors.Is(err, ErrConsumerNotFound)
 		lookupErr = err == ErrJetStreamNotEnabled || err == ErrTimeout || err == context.DeadlineExceeded
@@ -1563,6 +1574,19 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		if !(isPullMode && lookupErr && consumerBound) {
 			return nil, err
 		}
+	case skipCInfo:
+		// When skipping consumer info, need to rely on the manually passed sub options
+		// to match the expected behavior from the subscription.
+		hasFC, hbi = o.cfg.FlowControl, o.cfg.Heartbeat
+		hasHeartbeats = hbi > 0
+		maxap = o.cfg.MaxAckPending
+		deliver = o.cfg.DeliverSubject
+		if consumerBound {
+			break
+		}
+
+		// When not bound to a consumer already, proceed to create.
+		fallthrough
 	default:
 		// Attempt to create consumer if not found nor using Bind.
 		shouldCreate = true
@@ -1572,7 +1596,6 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 			deliver = nc.NewInbox()
 			cfg.DeliverSubject = deliver
 		}
-
 		// Do filtering always, server will clear as needed.
 		cfg.FilterSubject = subj
 
@@ -1610,6 +1633,8 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 	if isPullMode {
 		nms = fmt.Sprintf(js.apiSubj(apiRequestNextT), stream, consumer)
 		deliver = nc.NewInbox()
+		// for pull consumers, create a wildcard subscription to differentiate pull requests
+		deliver += ".*"
 	}
 
 	// In case this has a context, then create a child context that
@@ -1657,8 +1682,14 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 	}
 
 	// If we are creating or updating let's process that request.
+	consName := o.cfg.Name
 	if shouldCreate {
-		info, err := js.upsertConsumer(stream, cfg.Durable, ccreq.Config)
+		if cfg.Durable != "" {
+			consName = cfg.Durable
+		} else if consName == "" {
+			consName = getHash(nuid.Next())
+		}
+		info, err := js.upsertConsumer(stream, consName, ccreq.Config)
 		if err != nil {
 			var apiErr *APIError
 			if ok := errors.As(err, &apiErr); !ok {
@@ -1853,11 +1884,11 @@ func (sub *Subscription) checkOrderedMsgs(m *Msg) bool {
 	}
 
 	// Normal message here.
-	tokens, err := getMetadataFields(m.Reply)
+	tokens, err := parser.GetMetadataFields(m.Reply)
 	if err != nil {
 		return false
 	}
-	sseq, dseq := uint64(parseNum(tokens[ackStreamSeqTokenPos])), uint64(parseNum(tokens[ackConsumerSeqTokenPos]))
+	sseq, dseq := parser.ParseNum(tokens[ackStreamSeqTokenPos]), parser.ParseNum(tokens[ackConsumerSeqTokenPos])
 
 	jsi := sub.jsi
 	if dseq != jsi.dseq {
@@ -1961,40 +1992,22 @@ func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
 		cfg.DeliverPolicy = DeliverByStartSequencePolicy
 		cfg.OptStartSeq = sseq
 
-		ccSubj := fmt.Sprintf(apiLegacyConsumerCreateT, jsi.stream)
-		j, err := json.Marshal(jsi.ccreq)
 		js := jsi.js
 		sub.mu.Unlock()
 
+		consName := nuid.Next()
+		cinfo, err := js.upsertConsumer(jsi.stream, consName, cfg)
 		if err != nil {
-			pushErr(err)
-			return
-		}
-
-		resp, err := nc.Request(js.apiSubj(ccSubj), j, js.opts.wait)
-		if err != nil {
-			if errors.Is(err, ErrNoResponders) || errors.Is(err, ErrTimeout) {
+			var apiErr *APIError
+			if errors.Is(err, ErrJetStreamNotEnabled) || errors.Is(err, ErrTimeout) {
 				// if creating consumer failed, retry
 				return
-			}
-			pushErr(err)
-			return
-		}
-
-		var cinfo consumerResponse
-		err = json.Unmarshal(resp.Data, &cinfo)
-		if err != nil {
-			pushErr(err)
-			return
-		}
-
-		if cinfo.Error != nil {
-			if cinfo.Error.ErrorCode == JSErrCodeInsufficientResourcesErr {
+			} else if errors.As(err, &apiErr) && apiErr.ErrorCode == JSErrCodeInsufficientResourcesErr {
 				// retry for insufficient resources, as it may mean that client is connected to a running
 				// server in cluster while the server hosting R1 JetStream resources is restarting
 				return
 			}
-			pushErr(cinfo.Error)
+			pushErr(err)
 			return
 		}
 
@@ -2107,7 +2120,7 @@ func (nc *Conn) checkForSequenceMismatch(msg *Msg, s *Subscription, jsi *jsSub)
 		return
 	}
 
-	tokens, err := getMetadataFields(ctrl)
+	tokens, err := parser.GetMetadataFields(ctrl)
 	if err != nil {
 		return
 	}
@@ -2125,7 +2138,7 @@ func (nc *Conn) checkForSequenceMismatch(msg *Msg, s *Subscription, jsi *jsSub)
 	if ldseq != dseq {
 		// Dispatch async error including details such as
 		// from where the consumer could be restarted.
-		sseq := parseNum(tokens[ackStreamSeqTokenPos])
+		sseq := parser.ParseNum(tokens[ackStreamSeqTokenPos])
 		if ordered {
 			s.mu.Lock()
 			s.resetOrderedConsumer(jsi.sseq + 1)
@@ -2133,8 +2146,8 @@ func (nc *Conn) checkForSequenceMismatch(msg *Msg, s *Subscription, jsi *jsSub)
 		} else {
 			ecs := &ErrConsumerSequenceMismatch{
 				StreamResumeSequence: uint64(sseq),
-				ConsumerSequence:     uint64(parseNum(dseq)),
-				LastConsumerSequence: uint64(parseNum(ldseq)),
+				ConsumerSequence:     parser.ParseNum(dseq),
+				LastConsumerSequence: parser.ParseNum(ldseq),
 			}
 			nc.handleConsumerSequenceMismatch(s, ecs)
 		}
@@ -2163,6 +2176,22 @@ type subOpts struct {
 	// For an ordered consumer.
 	ordered bool
 	ctx     context.Context
+
+	// To disable calling ConsumerInfo
+	skipCInfo bool
+}
+
+// SkipConsumerLookup will omit lookipng up consumer when [Bind], [Durable]
+// or [ConsumerName] are provided.
+//
+// NOTE: This setting may cause an existing consumer to be overwritten. Also,
+// because consumer lookup is skipped, all consumer options like AckPolicy,
+// DeliverSubject etc. need to be provided even if consumer already exists.
+func SkipConsumerLookup() SubOpt {
+	return subOptFn(func(opts *subOpts) error {
+		opts.skipCInfo = true
+		return nil
+	})
 }
 
 // OrderedConsumer will create a FIFO direct/ephemeral consumer for in order delivery of messages.
@@ -2192,7 +2221,7 @@ func Description(description string) SubOpt {
 }
 
 // Durable defines the consumer name for JetStream durable subscribers.
-// This function will return ErrInvalidConsumerName in the name contains
+// This function will return ErrInvalidConsumerName if the name contains
 // any dot ".".
 func Durable(consumer string) SubOpt {
 	return subOptFn(func(opts *subOpts) error {
@@ -2486,6 +2515,14 @@ func ConsumerMemoryStorage() SubOpt {
 	})
 }
 
+// ConsumerName sets the name for a consumer.
+func ConsumerName(name string) SubOpt {
+	return subOptFn(func(opts *subOpts) error {
+		opts.cfg.Name = name
+		return nil
+	})
+}
+
 func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error) {
 	sub.mu.Lock()
 	// TODO(dlc) - Better way to mark especially if we attach.
@@ -2583,12 +2620,12 @@ func checkMsg(msg *Msg, checkSts, isNoWait bool) (usrMsg bool, err error) {
 			err = ErrTimeout
 		}
 	case jetStream409Sts:
-		if strings.Contains(strings.ToLower(string(msg.Header.Get(descrHdr))), "consumer deleted") {
+		if strings.Contains(strings.ToLower(msg.Header.Get(descrHdr)), "consumer deleted") {
 			err = ErrConsumerDeleted
 			break
 		}
 
-		if strings.Contains(strings.ToLower(string(msg.Header.Get(descrHdr))), "leadership change") {
+		if strings.Contains(strings.ToLower(msg.Header.Get(descrHdr)), "leadership change") {
 			err = ErrConsumerLeadershipChanged
 			break
 		}
@@ -2629,7 +2666,7 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 
 	nc := sub.conn
 	nms := sub.jsi.nms
-	rply := sub.jsi.deliver
+	rply, _ := newFetchInbox(jsi.deliver)
 	js := sub.jsi.js
 	pmc := len(sub.mch) > 0
 
@@ -2753,7 +2790,7 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 					err = sendReq()
 				} else if err == ErrTimeout && len(msgs) == 0 {
 					// If we get a 408, we will bail if we already collected some
-					// messages, otherwise ignore and go back calling NextMsg.
+					// messages, otherwise ignore and go back calling nextMsg.
 					err = nil
 				}
 			}
@@ -2766,6 +2803,28 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 	return msgs, nil
 }
 
+// newFetchInbox returns subject used as reply subject when sending pull requests
+// as well as request ID. For non-wildcard subject, request ID is empty and
+// passed subject is not transformed
+func newFetchInbox(subj string) (string, string) {
+	if !strings.HasSuffix(subj, ".*") {
+		return subj, ""
+	}
+	reqID := nuid.Next()
+	var sb strings.Builder
+	sb.WriteString(subj[:len(subj)-1])
+	sb.WriteString(reqID)
+	return sb.String(), reqID
+}
+
+func subjectMatchesReqID(subject, reqID string) bool {
+	subjectParts := strings.Split(subject, ".")
+	if len(subjectParts) < 2 {
+		return false
+	}
+	return subjectParts[len(subjectParts)-1] == reqID
+}
+
 // MessageBatch provides methods to retrieve messages consumed using [Subscribe.FetchBatch].
 type MessageBatch interface {
 	// Messages returns a channel on which messages will be published.
@@ -2835,7 +2894,7 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 
 	nc := sub.conn
 	nms := sub.jsi.nms
-	rply := sub.jsi.deliver
+	rply, reqID := newFetchInbox(sub.jsi.deliver)
 	js := sub.jsi.js
 	pmc := len(sub.mch) > 0
 
@@ -2964,6 +3023,10 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 			usrMsg, err = checkMsg(msg, true, false)
 			if err != nil {
 				if err == ErrTimeout {
+					if reqID != "" && !subjectMatchesReqID(msg.Subject, reqID) {
+						// ignore timeout message from server if it comes from a different pull request
+						continue
+					}
 					err = nil
 				}
 				break
@@ -3199,60 +3262,6 @@ const (
 	ackNumPendingTokenPos   = 10
 )
 
-func getMetadataFields(subject string) ([]string, error) {
-	const v1TokenCounts = 9
-	const v2TokenCounts = 12
-	const noDomainName = "_"
-
-	const btsep = '.'
-	tsa := [v2TokenCounts]string{}
-	start, tokens := 0, tsa[:0]
-	for i := 0; i < len(subject); i++ {
-		if subject[i] == btsep {
-			tokens = append(tokens, subject[start:i])
-			start = i + 1
-		}
-	}
-	tokens = append(tokens, subject[start:])
-	//
-	// Newer server will include the domain name and account hash in the subject,
-	// and a token at the end.
-	//
-	// Old subject was:
-	// $JS.ACK.<stream>.<consumer>.<delivered>.<sseq>.<cseq>.<tm>.<pending>
-	//
-	// New subject would be:
-	// $JS.ACK.<domain>.<account hash>.<stream>.<consumer>.<delivered>.<sseq>.<cseq>.<tm>.<pending>.<a token with a random value>
-	//
-	// v1 has 9 tokens, v2 has 12, but we must not be strict on the 12th since
-	// it may be removed in the future. Also, the library has no use for it.
-	// The point is that a v2 ACK subject is valid if it has at least 11 tokens.
-	//
-	l := len(tokens)
-	// If lower than 9 or more than 9 but less than 11, report an error
-	if l < v1TokenCounts || (l > v1TokenCounts && l < v2TokenCounts-1) {
-		return nil, ErrNotJSMessage
-	}
-	if tokens[0] != "$JS" || tokens[1] != "ACK" {
-		return nil, ErrNotJSMessage
-	}
-	// For v1 style, we insert 2 empty tokens (domain and hash) so that the
-	// rest of the library references known fields at a constant location.
-	if l == 9 {
-		// Extend the array (we know the backend is big enough)
-		tokens = append(tokens, _EMPTY_, _EMPTY_)
-		// Move to the right anything that is after "ACK" token.
-		copy(tokens[ackDomainTokenPos+2:], tokens[ackDomainTokenPos:])
-		// Clear the domain and hash tokens
-		tokens[ackDomainTokenPos], tokens[ackAccHashTokenPos] = _EMPTY_, _EMPTY_
-
-	} else if tokens[ackDomainTokenPos] == noDomainName {
-		// If domain is "_", replace with empty value.
-		tokens[ackDomainTokenPos] = _EMPTY_
-	}
-	return tokens, nil
-}
-
 // Metadata retrieves the metadata from a JetStream message. This method will
 // return an error for non-JetStream Msgs.
 func (m *Msg) Metadata() (*MsgMetadata, error) {
@@ -3260,45 +3269,24 @@ func (m *Msg) Metadata() (*MsgMetadata, error) {
 		return nil, err
 	}
 
-	tokens, err := getMetadataFields(m.Reply)
+	tokens, err := parser.GetMetadataFields(m.Reply)
 	if err != nil {
 		return nil, err
 	}
 
 	meta := &MsgMetadata{
 		Domain:       tokens[ackDomainTokenPos],
-		NumDelivered: uint64(parseNum(tokens[ackNumDeliveredTokenPos])),
-		NumPending:   uint64(parseNum(tokens[ackNumPendingTokenPos])),
-		Timestamp:    time.Unix(0, parseNum(tokens[ackTimestampSeqTokenPos])),
+		NumDelivered: parser.ParseNum(tokens[ackNumDeliveredTokenPos]),
+		NumPending:   parser.ParseNum(tokens[ackNumPendingTokenPos]),
+		Timestamp:    time.Unix(0, int64(parser.ParseNum(tokens[ackTimestampSeqTokenPos]))),
 		Stream:       tokens[ackStreamTokenPos],
 		Consumer:     tokens[ackConsumerTokenPos],
 	}
-	meta.Sequence.Stream = uint64(parseNum(tokens[ackStreamSeqTokenPos]))
-	meta.Sequence.Consumer = uint64(parseNum(tokens[ackConsumerSeqTokenPos]))
+	meta.Sequence.Stream = parser.ParseNum(tokens[ackStreamSeqTokenPos])
+	meta.Sequence.Consumer = parser.ParseNum(tokens[ackConsumerSeqTokenPos])
 	return meta, nil
 }
 
-// Quick parser for positive numbers in ack reply encoding.
-func parseNum(d string) (n int64) {
-	if len(d) == 0 {
-		return -1
-	}
-
-	// ASCII numbers 0-9
-	const (
-		asciiZero = 48
-		asciiNine = 57
-	)
-
-	for _, dec := range d {
-		if dec < asciiZero || dec > asciiNine {
-			return -1
-		}
-		n = n*10 + (int64(dec) - asciiZero)
-	}
-	return n
-}
-
 // AckPolicy determines how the consumer should acknowledge delivered messages.
 type AckPolicy int
 
@@ -3632,3 +3620,17 @@ func (st *StorageType) UnmarshalJSON(data []byte) error {
 	}
 	return nil
 }
+
+// Length of our hash used for named consumers.
+const nameHashLen = 8
+
+// Computes a hash for the given `name`.
+func getHash(name string) string {
+	sha := sha256.New()
+	sha.Write([]byte(name))
+	b := sha.Sum(nil)
+	for i := 0; i < nameHashLen; i++ {
+		b[i] = rdigits[int(b[i]%base)]
+	}
+	return string(b[:nameHashLen])
+}

vendor/github.com/nats-io/nats.go/jsm.go

@@ -1110,7 +1110,7 @@ func (js *js) getMsg(name string, mreq *apiMsgGetRequest, opts ...JSOpt) (*RawSt
 
 	var hdr Header
 	if len(msg.Header) > 0 {
-		hdr, err = decodeHeadersMsg(msg.Header)
+		hdr, err = DecodeHeadersMsg(msg.Header)
 		if err != nil {
 			return nil, err
 		}

vendor/github.com/nats-io/nats.go/kv.go

@@ -23,6 +23,8 @@ import (
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/nats-io/nats.go/internal/parser"
 )
 
 // KeyValueManager is used to manage KeyValue stores.
@@ -889,7 +891,7 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 	w := &watcher{updates: make(chan KeyValueEntry, 256), ctx: o.ctx}
 
 	update := func(m *Msg) {
-		tokens, err := getMetadataFields(m.Reply)
+		tokens, err := parser.GetMetadataFields(m.Reply)
 		if err != nil {
 			return
 		}
@@ -907,7 +909,7 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 				op = KeyValuePurge
 			}
 		}
-		delta := uint64(parseNum(tokens[ackNumPendingTokenPos]))
+		delta := parser.ParseNum(tokens[ackNumPendingTokenPos])
 		w.mu.Lock()
 		defer w.mu.Unlock()
 		if !o.ignoreDeletes || (op != KeyValueDelete && op != KeyValuePurge) {
@@ -915,8 +917,8 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 				bucket:   kv.name,
 				key:      subj,
 				value:    m.Data,
-				revision: uint64(parseNum(tokens[ackStreamSeqTokenPos])),
-				created:  time.Unix(0, parseNum(tokens[ackTimestampSeqTokenPos])),
+				revision: parser.ParseNum(tokens[ackStreamSeqTokenPos]),
+				created:  time.Unix(0, int64(parser.ParseNum(tokens[ackTimestampSeqTokenPos]))),
 				delta:    delta,
 				op:       op,
 			}

vendor/github.com/nats-io/nats.go/nats.go

@@ -47,7 +47,7 @@ import (
 
 // Default Constants
 const (
-	Version                   = "1.24.0"
+	Version                   = "1.27.0"
 	DefaultURL                = "nats://127.0.0.1:4222"
 	DefaultPort               = 4222
 	DefaultMaxReconnect       = 60
@@ -211,6 +211,13 @@ type ErrHandler func(*Conn, *Subscription, error)
 // JWT for this user.
 type UserJWTHandler func() (string, error)
 
+// TLSCertHandler is used to fetch and return tls certificate.
+type TLSCertHandler func() (tls.Certificate, error)
+
+// RootCAsHandler is used to fetch and return a set of root certificate
+// authorities that clients use when verifying server certificates.
+type RootCAsHandler func() (*x509.CertPool, error)
+
 // SignatureHandler is used to sign a nonce from the server while
 // authenticating with nkeys. The user should sign the nonce and
 // return the raw signature. The client will base64 encode this to
@@ -299,6 +306,13 @@ type Options struct {
 	// transports.
 	TLSConfig *tls.Config
 
+	// TLSCertCB is used to fetch and return custom tls certificate.
+	TLSCertCB TLSCertHandler
+
+	// RootCAsCB is used to fetch and return a set of root certificate
+	// authorities that clients use when verifying server certificates.
+	RootCAsCB RootCAsHandler
+
 	// AllowReconnect enables reconnection logic to be used when we
 	// encounter a disconnect from the current server.
 	AllowReconnect bool
@@ -510,31 +524,32 @@ type Conn struct {
 	mu sync.RWMutex
 	// Opts holds the configuration of the Conn.
 	// Modifying the configuration of a running Conn is a race.
-	Opts    Options
-	wg      sync.WaitGroup
-	srvPool []*srv
-	current *srv
-	urls    map[string]struct{} // Keep track of all known URLs (used by processInfo)
-	conn    net.Conn
-	bw      *natsWriter
-	br      *natsReader
-	fch     chan struct{}
-	info    serverInfo
-	ssid    int64
-	subsMu  sync.RWMutex
-	subs    map[int64]*Subscription
-	ach     *asyncCallbacksHandler
-	pongs   []chan struct{}
-	scratch [scratchSize]byte
-	status  Status
-	initc   bool // true if the connection is performing the initial connect
-	err     error
-	ps      *parseState
-	ptmr    *time.Timer
-	pout    int
-	ar      bool // abort reconnect
-	rqch    chan struct{}
-	ws      bool // true if a websocket connection
+	Opts          Options
+	wg            sync.WaitGroup
+	srvPool       []*srv
+	current       *srv
+	urls          map[string]struct{} // Keep track of all known URLs (used by processInfo)
+	conn          net.Conn
+	bw            *natsWriter
+	br            *natsReader
+	fch           chan struct{}
+	info          serverInfo
+	ssid          int64
+	subsMu        sync.RWMutex
+	subs          map[int64]*Subscription
+	ach           *asyncCallbacksHandler
+	pongs         []chan struct{}
+	scratch       [scratchSize]byte
+	status        Status
+	statListeners map[Status][]chan Status
+	initc         bool // true if the connection is performing the initial connect
+	err           error
+	ps            *parseState
+	ptmr          *time.Timer
+	pout          int
+	ar            bool // abort reconnect
+	rqch          chan struct{}
+	ws            bool // true if a websocket connection
 
 	// New style response handler
 	respSub       string               // The wildcard subject
@@ -672,6 +687,15 @@ func (m *Msg) Equal(msg *Msg) bool {
 	return true
 }
 
+// Size returns a message size in bytes.
+func (m *Msg) Size() int {
+	if m.wsz != 0 {
+		return m.wsz
+	}
+	hdr, _ := m.headerBytes()
+	return len(m.Subject) + len(m.Reply) + len(hdr) + len(m.Data)
+}
+
 func (m *Msg) headerBytes() ([]byte, error) {
 	var hdr []byte
 	if len(m.Header) == 0 {
@@ -834,21 +858,27 @@ func Secure(tls ...*tls.Config) Option {
 // If Secure is not already set this will set it as well.
 func RootCAs(file ...string) Option {
 	return func(o *Options) error {
-		pool := x509.NewCertPool()
-		for _, f := range file {
-			rootPEM, err := os.ReadFile(f)
-			if err != nil || rootPEM == nil {
-				return fmt.Errorf("nats: error loading or parsing rootCA file: %w", err)
-			}
-			ok := pool.AppendCertsFromPEM(rootPEM)
-			if !ok {
-				return fmt.Errorf("nats: failed to parse root certificate from %q", f)
+		rootCAsCB := func() (*x509.CertPool, error) {
+			pool := x509.NewCertPool()
+			for _, f := range file {
+				rootPEM, err := os.ReadFile(f)
+				if err != nil || rootPEM == nil {
+					return nil, fmt.Errorf("nats: error loading or parsing rootCA file: %w", err)
+				}
+				ok := pool.AppendCertsFromPEM(rootPEM)
+				if !ok {
+					return nil, fmt.Errorf("nats: failed to parse root certificate from %q", f)
+				}
 			}
+			return pool, nil
 		}
 		if o.TLSConfig == nil {
 			o.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
 		}
-		o.TLSConfig.RootCAs = pool
+		if _, err := rootCAsCB(); err != nil {
+			return err
+		}
+		o.RootCAsCB = rootCAsCB
 		o.Secure = true
 		return nil
 	}
@@ -858,18 +888,24 @@ func RootCAs(file ...string) Option {
 // If Secure is not already set this will set it as well.
 func ClientCert(certFile, keyFile string) Option {
 	return func(o *Options) error {
-		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
-		if err != nil {
-			return fmt.Errorf("nats: error loading client certificate: %w", err)
-		}
-		cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
-		if err != nil {
-			return fmt.Errorf("nats: error parsing client certificate: %w", err)
+		tlsCertCB := func() (tls.Certificate, error) {
+			cert, err := tls.LoadX509KeyPair(certFile, keyFile)
+			if err != nil {
+				return tls.Certificate{}, fmt.Errorf("nats: error loading client certificate: %w", err)
+			}
+			cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
+			if err != nil {
+				return tls.Certificate{}, fmt.Errorf("nats: error parsing client certificate: %w", err)
+			}
+			return cert, nil
 		}
 		if o.TLSConfig == nil {
 			o.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
 		}
-		o.TLSConfig.Certificates = []tls.Certificate{cert}
+		if _, err := tlsCertCB(); err != nil {
+			return err
+		}
+		o.TLSCertCB = tlsCertCB
 		o.Secure = true
 		return nil
 	}
@@ -1969,11 +2005,23 @@ func (nc *Conn) makeTLSConn() error {
 		}
 	}
 	// Allow the user to configure their own tls.Config structure.
-	var tlsCopy *tls.Config
+	tlsCopy := &tls.Config{}
 	if nc.Opts.TLSConfig != nil {
 		tlsCopy = util.CloneTLSConfig(nc.Opts.TLSConfig)
-	} else {
-		tlsCopy = &tls.Config{}
+	}
+	if nc.Opts.TLSCertCB != nil {
+		cert, err := nc.Opts.TLSCertCB()
+		if err != nil {
+			return err
+		}
+		tlsCopy.Certificates = []tls.Certificate{cert}
+	}
+	if nc.Opts.RootCAsCB != nil {
+		rootCAs, err := nc.Opts.RootCAsCB()
+		if err != nil {
+			return err
+		}
+		tlsCopy.RootCAs = rootCAs
 	}
 	// If its blank we will override it with the current host
 	if tlsCopy.ServerName == _EMPTY_ {
@@ -2181,7 +2229,7 @@ func (nc *Conn) processConnectInit() error {
 	defer nc.conn.SetDeadline(time.Time{})
 
 	// Set our status to connecting.
-	nc.status = CONNECTING
+	nc.changeConnStatus(CONNECTING)
 
 	// Process the INFO protocol received from the server
 	err := nc.processExpectedInfo()
@@ -2273,7 +2321,7 @@ func (nc *Conn) connect() (bool, error) {
 		nc.initc = false
 	} else if nc.Opts.RetryOnFailedConnect {
 		nc.setup()
-		nc.status = RECONNECTING
+		nc.changeConnStatus(RECONNECTING)
 		nc.bw.switchToPending()
 		go nc.doReconnect(ErrNoServers)
 		err = nil
@@ -2466,6 +2514,9 @@ func (nc *Conn) sendConnect() error {
 	// reading byte-by-byte here is ok.
 	proto, err := nc.readProto()
 	if err != nil {
+		if !nc.initc && nc.Opts.AsyncErrorCB != nil {
+			nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, err) })
+		}
 		return err
 	}
 
@@ -2474,6 +2525,9 @@ func (nc *Conn) sendConnect() error {
 		// Read the rest now...
 		proto, err = nc.readProto()
 		if err != nil {
+			if !nc.initc && nc.Opts.AsyncErrorCB != nil {
+				nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, err) })
+			}
 			return err
 		}
 	}
@@ -2507,7 +2561,7 @@ func (nc *Conn) sendConnect() error {
 	}
 
 	// This is where we are truly connected.
-	nc.status = CONNECTED
+	nc.changeConnStatus(CONNECTED)
 
 	return nil
 }
@@ -2682,7 +2736,7 @@ func (nc *Conn) doReconnect(err error) {
 			if nc.ar {
 				break
 			}
-			nc.status = RECONNECTING
+			nc.changeConnStatus(RECONNECTING)
 			continue
 		}
 
@@ -2700,7 +2754,7 @@ func (nc *Conn) doReconnect(err error) {
 		// Now send off and clear pending buffer
 		nc.err = nc.flushReconnectPendingItems()
 		if nc.err != nil {
-			nc.status = RECONNECTING
+			nc.changeConnStatus(RECONNECTING)
 			// Stop the ping timer (if set)
 			nc.stopPingTimer()
 			// Since processConnectInit() returned without error, the
@@ -2753,7 +2807,7 @@ func (nc *Conn) processOpErr(err error) {
 
 	if nc.Opts.AllowReconnect && nc.status == CONNECTED {
 		// Set our new status
-		nc.status = RECONNECTING
+		nc.changeConnStatus(RECONNECTING)
 		// Stop ping timer if set
 		nc.stopPingTimer()
 		if nc.conn != nil {
@@ -2772,7 +2826,7 @@ func (nc *Conn) processOpErr(err error) {
 		return
 	}
 
-	nc.status = DISCONNECTED
+	nc.changeConnStatus(DISCONNECTED)
 	nc.err = err
 	nc.mu.Unlock()
 	nc.close(CLOSED, true, nil)
@@ -3049,7 +3103,7 @@ func (nc *Conn) processMsg(data []byte) {
 	if nc.ps.ma.hdr > 0 {
 		hbuf := msgPayload[:nc.ps.ma.hdr]
 		msgPayload = msgPayload[nc.ps.ma.hdr:]
-		h, err = decodeHeadersMsg(hbuf)
+		h, err = DecodeHeadersMsg(hbuf)
 		if err != nil {
 			// We will pass the message through but send async error.
 			nc.mu.Lock()
@@ -3564,8 +3618,8 @@ const (
 	statusLen          = 3 // e.g. 20x, 40x, 50x
 )
 
-// decodeHeadersMsg will decode and headers.
-func decodeHeadersMsg(data []byte) (Header, error) {
+// DecodeHeadersMsg will decode and headers.
+func DecodeHeadersMsg(data []byte) (Header, error) {
 	br := bufio.NewReaderSize(bytes.NewReader(data), 128)
 	tp := textproto.NewReader(br)
 	l, err := tp.ReadLine()
@@ -5021,15 +5075,15 @@ func (nc *Conn) close(status Status, doCBs bool, err error) {
 	nc.subs = nil
 	nc.subsMu.Unlock()
 
-	nc.status = status
+	nc.changeConnStatus(status)
 
 	// Perform appropriate callback if needed for a disconnect.
 	if doCBs {
 		if nc.conn != nil {
-			if nc.Opts.DisconnectedErrCB != nil {
-				nc.ach.push(func() { nc.Opts.DisconnectedErrCB(nc, err) })
-			} else if nc.Opts.DisconnectedCB != nil {
-				nc.ach.push(func() { nc.Opts.DisconnectedCB(nc) })
+			if disconnectedErrCB := nc.Opts.DisconnectedErrCB; disconnectedErrCB != nil {
+				nc.ach.push(func() { disconnectedErrCB(nc, err) })
+			} else if disconnectedCB := nc.Opts.DisconnectedCB; disconnectedCB != nil {
+				nc.ach.push(func() { disconnectedCB(nc) })
 			}
 		}
 		if nc.Opts.ClosedCB != nil {
@@ -5166,7 +5220,7 @@ func (nc *Conn) drainConnection() {
 
 	// Flip State
 	nc.mu.Lock()
-	nc.status = DRAINING_PUBS
+	nc.changeConnStatus(DRAINING_PUBS)
 	nc.mu.Unlock()
 
 	// Do publish drain via Flush() call.
@@ -5201,7 +5255,7 @@ func (nc *Conn) Drain() error {
 		nc.mu.Unlock()
 		return nil
 	}
-	nc.status = DRAINING_SUBS
+	nc.changeConnStatus(DRAINING_SUBS)
 	go nc.drainConnection()
 	nc.mu.Unlock()
 
@@ -5411,6 +5465,68 @@ func (nc *Conn) GetClientID() (uint64, error) {
 	return nc.info.CID, nil
 }
 
+// StatusChanged returns a channel on which given list of connection status changes will be reported.
+// If no statuses are provided, defaults will be used: CONNECTED, RECONNECTING, DISCONNECTED, CLOSED.
+func (nc *Conn) StatusChanged(statuses ...Status) chan Status {
+	if len(statuses) == 0 {
+		statuses = []Status{CONNECTED, RECONNECTING, DISCONNECTED, CLOSED}
+	}
+	ch := make(chan Status)
+	for _, s := range statuses {
+		nc.registerStatusChangeListener(s, ch)
+	}
+	return ch
+}
+
+// registerStatusChangeListener registers a channel waiting for a specific status change event.
+// Status change events are non-blocking - if no receiver is waiting for the status change,
+// it will not be sent on the channel. Closed channels are ignored.
+func (nc *Conn) registerStatusChangeListener(status Status, ch chan Status) {
+	nc.mu.Lock()
+	defer nc.mu.Unlock()
+	if nc.statListeners == nil {
+		nc.statListeners = make(map[Status][]chan Status)
+	}
+	if _, ok := nc.statListeners[status]; !ok {
+		nc.statListeners[status] = make([]chan Status, 0)
+	}
+	nc.statListeners[status] = append(nc.statListeners[status], ch)
+}
+
+// sendStatusEvent sends connection status event to all channels.
+// If channel is closed, or there is no listener, sendStatusEvent
+// will not block. Lock should be held entering.
+func (nc *Conn) sendStatusEvent(s Status) {
+Loop:
+	for i := 0; i < len(nc.statListeners[s]); i++ {
+		// make sure channel is not closed
+		select {
+		case <-nc.statListeners[s][i]:
+			// if chan is closed, remove it
+			nc.statListeners[s][i] = nc.statListeners[s][len(nc.statListeners[s])-1]
+			nc.statListeners[s] = nc.statListeners[s][:len(nc.statListeners[s])-1]
+			i--
+			continue Loop
+		default:
+		}
+		// only send event if someone's listening
+		select {
+		case nc.statListeners[s][i] <- s:
+		default:
+		}
+	}
+}
+
+// changeConnStatus changes connections status and sends events
+// to all listeners. Lock should be held entering.
+func (nc *Conn) changeConnStatus(status Status) {
+	if nc == nil {
+		return
+	}
+	nc.sendStatusEvent(status)
+	nc.status = status
+}
+
 // NkeyOptionFromSeed will load an nkey pair from a seed file.
 // It will return the NKey Option and will handle
 // signing of nonce challenges from the server. It will take

vendor/github.com/nats-io/nats.go/object.go

@@ -29,6 +29,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/nats-io/nats.go/internal/parser"
 	"github.com/nats-io/nuid"
 )
 
@@ -368,14 +369,23 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 		return perr
 	}
 
-	purgePartial := func() { obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: chunkSubj}) }
-
 	// Create our own JS context to handle errors etc.
-	js, err := obs.js.nc.JetStream(PublishAsyncErrHandler(func(js JetStream, _ *Msg, err error) { setErr(err) }))
+	jetStream, err := obs.js.nc.JetStream(PublishAsyncErrHandler(func(js JetStream, _ *Msg, err error) { setErr(err) }))
 	if err != nil {
 		return nil, err
 	}
 
+	defer jetStream.(*js).cleanupReplySub()
+
+	purgePartial := func() {
+		// wait until all pubs are complete or up to default timeout before attempting purge
+		select {
+		case <-jetStream.PublishAsyncComplete():
+		case <-time.After(obs.js.opts.wait):
+		}
+		obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: chunkSubj})
+	}
+
 	m, h := NewMsg(chunkSubj), sha256.New()
 	chunk, sent, total := make([]byte, meta.Opts.ChunkSize), 0, uint64(0)
 
@@ -416,7 +426,7 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 			h.Write(m.Data)
 
 			// Send msg itself.
-			if _, err := js.PublishMsgAsync(m); err != nil {
+			if _, err := jetStream.PublishMsgAsync(m); err != nil {
 				purgePartial()
 				return nil, err
 			}
@@ -451,7 +461,7 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 	}
 
 	// Publish the meta message.
-	_, err = js.PublishMsgAsync(mm)
+	_, err = jetStream.PublishMsgAsync(mm)
 	if err != nil {
 		if r != nil {
 			purgePartial()
@@ -461,7 +471,7 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 
 	// Wait for all to be processed.
 	select {
-	case <-js.PublishAsyncComplete():
+	case <-jetStream.PublishAsyncComplete():
 		if err := getErr(); err != nil {
 			if r != nil {
 				purgePartial()
@@ -628,7 +638,7 @@ func (obs *obs) Get(name string, opts ...GetObjectOpt) (ObjectResult, error) {
 			}
 		}
 
-		tokens, err := getMetadataFields(m.Reply)
+		tokens, err := parser.GetMetadataFields(m.Reply)
 		if err != nil {
 			gotErr(m, err)
 			return
@@ -1207,7 +1217,7 @@ func (o *objResult) Read(p []byte) (n int, err error) {
 		}
 	}
 	if o.err != nil {
-		return 0, err
+		return 0, o.err
 	}
 	if o.r == nil {
 		return 0, io.EOF

vendor/github.com/nats-io/nats.go/parser.go

@@ -1,4 +1,4 @@
-// Copyright 2012-2122 The NATS Authors
+// Copyright 2012-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at

vendor/github.com/nats-io/nats.go/ws.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 The NATS Authors
+// Copyright 2021-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -16,7 +16,6 @@ package nats
 import (
 	"bufio"
 	"bytes"
-	"compress/flate"
 	"crypto/rand"
 	"crypto/sha1"
 	"encoding/base64"
@@ -30,6 +29,8 @@ import (
 	"strings"
 	"time"
 	"unicode/utf8"
+
+	"github.com/klauspost/compress/flate"
 )
 
 type wsOpCode int
@@ -448,8 +449,12 @@ func (w *websocketWriter) Write(p []byte) (int, error) {
 			} else {
 				w.compressor.Reset(buf)
 			}
-			w.compressor.Write(p)
-			w.compressor.Close()
+			if n, err = w.compressor.Write(p); err != nil {
+				return n, err
+			}
+			if err = w.compressor.Flush(); err != nil {
+				return n, err
+			}
 			b := buf.Bytes()
 			p = b[:len(b)-4]
 		}
@@ -550,7 +555,7 @@ func wsFillFrameHeader(fh []byte, compressed bool, frameType wsOpCode, l int) (i
 
 func (nc *Conn) wsInitHandshake(u *url.URL) error {
 	compress := nc.Opts.Compression
-	tlsRequired := u.Scheme == wsSchemeTLS || nc.Opts.Secure || nc.Opts.TLSConfig != nil
+	tlsRequired := u.Scheme == wsSchemeTLS || nc.Opts.Secure || nc.Opts.TLSConfig != nil || nc.Opts.TLSCertCB != nil || nc.Opts.RootCAsCB != nil
 	// Do TLS here as needed.
 	if tlsRequired {
 		if err := nc.makeTLSConn(); err != nil {
@@ -692,6 +697,9 @@ func (nc *Conn) wsEnqueueCloseMsgLocked(status int, payload string) {
 	wr.cm = frame
 	wr.cmDone = true
 	nc.bw.flush()
+	if c := wr.compressor; c != nil {
+		c.Close()
+	}
 }
 
 func (nc *Conn) wsEnqueueControlMsg(needsLock bool, frameType wsOpCode, payload []byte) {

vendor/modules.txt

@@ -1325,7 +1325,7 @@ github.com/mschoch/smat
 # github.com/nats-io/jwt/v2 v2.4.1
 ## explicit; go 1.18
 github.com/nats-io/jwt/v2
-# github.com/nats-io/nats-server/v2 v2.9.17
+# github.com/nats-io/nats-server/v2 v2.9.19
 ## explicit; go 1.19
 github.com/nats-io/nats-server/v2/conf
 github.com/nats-io/nats-server/v2/internal/ldap
@@ -1333,10 +1333,11 @@ github.com/nats-io/nats-server/v2/logger
 github.com/nats-io/nats-server/v2/server
 github.com/nats-io/nats-server/v2/server/pse
 github.com/nats-io/nats-server/v2/server/sysmem
-# github.com/nats-io/nats.go v1.24.0
+# github.com/nats-io/nats.go v1.27.0
 ## explicit; go 1.19
 github.com/nats-io/nats.go
 github.com/nats-io/nats.go/encoders/builtin
+github.com/nats-io/nats.go/internal/parser
 github.com/nats-io/nats.go/util
 # github.com/nats-io/nkeys v0.4.4
 ## explicit; go 1.19