Repository: nats-io/nats.go
Pull request #1705 (2024-08-27 11:41:30)

[FIXED] Invalid fetch sequence in ordered consumer Fetch and Next after timeout
Signed-off-by: Piotr Piotrowski <piotr@synadia.com>


Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
new file mode 100644
index 0000000..0ea16de
--- /dev/null
+++ b/.github/workflows/ci.yaml
@@ -0,0 +1,74 @@
+name: ci
+on: 
+  push:
+    branches:
+      - main
+  pull_request:
+  release:
+    types: [published]
+
+jobs:
+    lint:
+        runs-on: ubuntu-latest
+        steps:
+            - name: Checkout code
+              uses: actions/checkout@v4
+      
+            - name: Setup Go
+              uses: actions/setup-go@v5
+              with:
+                go-version: 'stable'
+
+            - name: Install deps
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go get -t ./...
+                go install honnef.co/go/tools/cmd/staticcheck@latest
+                go install github.com/client9/misspell/cmd/misspell@latest
+                go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest
+            
+            - name: Run linters
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                $(exit $(go fmt -modfile=go_test.mod  ./... | wc -l))
+                go vet -modfile=go_test.mod ./... 
+                GOFLAGS="-mod=mod -modfile=go_test.mod" staticcheck ./...
+                find . -type f -name "*.go" | xargs misspell -error -locale US
+                golangci-lint run --timeout 5m0s ./jetstream/...
+
+    test:
+        runs-on: ubuntu-latest
+
+        strategy:
+            matrix:
+              go: [ "1.22", "1.23" ]
+        steps:
+            - name: Checkout code
+              uses: actions/checkout@v4
+      
+            - name: Setup Go
+              uses: actions/setup-go@v5
+              with:
+                go-version: ${{ matrix.go }}
+
+            - name: Install deps
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go install github.com/mattn/goveralls@latest
+                go install github.com/wadey/gocovmerge@latest
+
+            - name: Test and coverage
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
+                if [ "${{ matrix.go }}" = "1.23" ]; then
+                  ./scripts/cov.sh CI
+                else
+                  go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing
+                fi
+
+            - name: Coveralls
+              if: matrix.go == '1.23'
+              uses: coverallsapp/github-action@v2
+              with:
+               file: acc.out
\ No newline at end of file
diff --git a/.github/workflows/dependencies.yaml b/.github/workflows/dependencies.yaml
new file mode 100644
index 0000000..5d9a7dc
--- /dev/null
+++ b/.github/workflows/dependencies.yaml
@@ -0,0 +1,61 @@
+name: License Check
+
+on:
+  push:
+    paths:
+      - 'go.mod'
+    branches:
+      - main
+
+jobs:
+  license-check:
+    runs-on: ubuntu-latest
+
+    env:
+      BRANCH_NAME: update-report-branch-${{ github.run_id }}
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0  # Fetch all history for all branches and tags
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.22'
+
+      - name: Install go-licenses
+        run: go install github.com/google/go-licenses@latest
+        # We need this step because of test dependencies and how they are handled in nats.go
+      - name: Run go mod tidy
+        run: go mod tidy
+      - name: Run license check
+        run: go-licenses report ./... --template dependencies.tpl > dependencies.md
+
+      - name: Configure git
+        run: |
+          git config user.name 'github-actions[bot]'
+          git config user.email 'github-actions[bot]@users.noreply.github.com'
+
+      - name: Check for changes
+        id: git_diff
+        run: |
+          git fetch
+          git diff --exit-code dependencies.md || echo "has_changes=true" >> $GITHUB_ENV
+
+      - name: Commit changes
+        if: env.has_changes == 'true'
+        run: |
+          git checkout -b "$BRANCH_NAME"
+          git add dependencies.md
+          git commit -m "Update dependencies.md"
+          git push -u origin "$BRANCH_NAME"
+
+      - name: Create Pull Request
+        if: env.has_changes == 'true'
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          gh pr create --title "Update dependencies.md" --body "This PR updates the dependencies report" --head "$BRANCH_NAME" --base main
+
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 9a6b4a8..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,36 +0,0 @@
-language: go
-go:
-- "1.22.x"
-- "1.21.x"
-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.22 ]]; then
-    go install github.com/mattn/goveralls@latest;
-    go install github.com/wadey/gocovmerge@latest;
-    go install honnef.co/go/tools/cmd/staticcheck@latest;
-    go install github.com/client9/misspell/cmd/misspell@latest;
-  fi
-before_script:
-- $(exit $(go fmt ./... | wc -l))
-- go vet -modfile=go_test.mod ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.22 ]]; then
-    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.22 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing; fi
-after_success:
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.22 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
-
-jobs:
-  include:
-  - name: "Go: 1.22.x (nats-server@main)"
-    go: "1.22.x"
-    before_script:
-    - go get -modfile go_test.mod github.com/nats-io/nats-server/v2@main
-  allow_failures:
-  - name: "Go: 1.22.x (nats-server@main)"
diff --git a/README.md b/README.md
index 0d8ccca..84accda 100644
--- a/README.md
+++ b/README.md
@@ -7,8 +7,8 @@ A [Go](http://golang.org) client for the [NATS messaging system](https://nats.io
 [License-Image]: https://img.shields.io/badge/License-Apache2-blue.svg
 [ReportCard-Url]: https://goreportcard.com/report/github.com/nats-io/nats.go
 [ReportCard-Image]: https://goreportcard.com/badge/github.com/nats-io/nats.go
-[Build-Status-Url]: https://travis-ci.com/github/nats-io/nats.go
-[Build-Status-Image]: https://travis-ci.com/nats-io/nats.go.svg?branch=main
+[Build-Status-Url]: https://github.com/nats-io/nats.go/actions
+[Build-Status-Image]: https://github.com/nats-io/nats.go/actions/workflows/ci.yaml/badge.svg?branch=main
 [GoDoc-Url]: https://pkg.go.dev/github.com/nats-io/nats.go
 [GoDoc-Image]: https://img.shields.io/badge/GoDoc-reference-007d9c
 [Coverage-Url]: https://coveralls.io/r/nats-io/nats.go?branch=main
diff --git a/dependencies.tpl b/dependencies.tpl
new file mode 100644
index 0000000..5d01380
--- /dev/null
+++ b/dependencies.tpl
@@ -0,0 +1,8 @@
+# External Dependencies
+
+This file lists the dependencies used in this repository.
+
+| Dependency                                       | License                                 |
+|--------------------------------------------------|-----------------------------------------|
+{{ range . }}| {{.Name}}                           | {{.LicenseName}}                        |
+{{ end }}
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index e401cb9..3b94146 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -122,7 +122,8 @@ type (
 	// to operate on a stream.
 	StreamManager interface {
 		// CreateStream creates a new stream with given config and returns an
-		// interface to operate on it. If stream with given name already exists,
+		// interface to operate on it. If stream with given name already exists
+		// and its configuration differs from the provided one,
 		// ErrStreamNameAlreadyInUse is returned.
 		CreateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
 
@@ -435,7 +436,7 @@ func NewWithAPIPrefix(nc *nats.Conn, apiPrefix string, opts ...JetStreamOpt) (Je
 		}
 	}
 	if apiPrefix == "" {
-		return nil, fmt.Errorf("API prefix cannot be empty")
+		return nil, errors.New("API prefix cannot be empty")
 	}
 	if !strings.HasSuffix(apiPrefix, ".") {
 		jsOpts.apiPrefix = fmt.Sprintf("%s.", apiPrefix)
diff --git a/jetstream/kv.go b/jetstream/kv.go
index ae892b4..91a1bdc 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -123,6 +123,7 @@ type (
 
 		// Update will update the value if the latest revision matches.
 		// If the provided revision is not the latest, Update will return an error.
+		// Update also resets the TTL associated with the key (if any).
 		Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)
 
 		// Delete will place a delete marker and leave all revisions. A history
diff --git a/jetstream/message.go b/jetstream/message.go
index 81e1512..095f139 100644
--- a/jetstream/message.go
+++ b/jetstream/message.go
@@ -16,6 +16,7 @@ package jetstream
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -434,7 +435,7 @@ func parsePending(msg *nats.Msg) (int, int, error) {
 	if msgsLeftStr != "" {
 		msgsLeft, err = strconv.Atoi(msgsLeftStr)
 		if err != nil {
-			return 0, 0, fmt.Errorf("nats: invalid format of Nats-Pending-Messages")
+			return 0, 0, errors.New("nats: invalid format of Nats-Pending-Messages")
 		}
 	}
 	bytesLeftStr := msg.Header.Get("Nats-Pending-Bytes")
@@ -442,7 +443,7 @@ func parsePending(msg *nats.Msg) (int, int, error) {
 	if bytesLeftStr != "" {
 		bytesLeft, err = strconv.Atoi(bytesLeftStr)
 		if err != nil {
-			return 0, 0, fmt.Errorf("nats: invalid format of Nats-Pending-Bytes")
+			return 0, 0, errors.New("nats: invalid format of Nats-Pending-Bytes")
 		}
 	}
 	return msgsLeft, bytesLeft, nil
diff --git a/jetstream/ordered.go b/jetstream/ordered.go
index 199b8c3..af598f2 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -402,7 +402,9 @@ func (c *orderedConsumer) Fetch(batch int, opts ...FetchOpt) (MessageBatch, erro
 			c.currentConsumer.Unlock()
 			return nil, ErrOrderedConsumerConcurrentRequests
 		}
-		c.cursor.streamSeq = c.runningFetch.sseq
+		if c.runningFetch.sseq != 0 {
+			c.cursor.streamSeq = c.runningFetch.sseq
+		}
 	}
 	c.currentConsumer.Unlock()
 	c.consumerType = consumerTypeFetch
@@ -438,7 +440,9 @@ func (c *orderedConsumer) FetchBytes(maxBytes int, opts ...FetchOpt) (MessageBat
 		if !c.runningFetch.done {
 			return nil, ErrOrderedConsumerConcurrentRequests
 		}
-		c.cursor.streamSeq = c.runningFetch.sseq
+		if c.runningFetch.sseq != 0 {
+			c.cursor.streamSeq = c.runningFetch.sseq
+		}
 	}
 	c.consumerType = consumerTypeFetch
 	sub := orderedSubscription{
@@ -747,7 +751,7 @@ func retryWithBackoff(f func(int) (bool, error), opts backoffOpts) error {
 	// if custom backoff is set, use it instead of other options
 	if len(opts.customBackoff) > 0 {
 		if opts.attempts != 0 {
-			return fmt.Errorf("cannot use custom backoff intervals when attempts are set")
+			return errors.New("cannot use custom backoff intervals when attempts are set")
 		}
 		for i, interval := range opts.customBackoff {
 			select {
@@ -774,7 +778,7 @@ func retryWithBackoff(f func(int) (bool, error), opts backoffOpts) error {
 		opts.maxInterval = 1 * time.Minute
 	}
 	if opts.attempts == 0 {
-		return fmt.Errorf("retry attempts have to be set when not using custom backoff intervals")
+		return errors.New("retry attempts have to be set when not using custom backoff intervals")
 	}
 	interval := opts.initialInterval
 	for i := 0; ; i++ {
diff --git a/jetstream/publish.go b/jetstream/publish.go
index d08cdf1..70e219a 100644
--- a/jetstream/publish.go
+++ b/jetstream/publish.go
@@ -81,6 +81,7 @@ type (
 		err        error
 		errCh      chan error
 		doneCh     chan *PubAck
+		reply      string
 	}
 
 	jetStreamClient struct {
@@ -280,17 +281,17 @@ func (js *jetStream) PublishMsgAsync(m *nats.Msg, opts ...PublishOpt) (PubAckFut
 	}
 
 	var id string
+	var reply string
 
 	// register new paf if not retrying
 	if paf == nil {
 		var err error
-		m.Reply, err = js.newAsyncReply()
-		defer func() { m.Reply = "" }()
+		reply, err = js.newAsyncReply()
 		if err != nil {
 			return nil, fmt.Errorf("nats: error creating async reply handler: %s", err)
 		}
-		id = m.Reply[js.replyPrefixLen:]
-		paf = &pubAckFuture{msg: m, jsClient: js.publisher, maxRetries: o.retryAttempts, retryWait: o.retryWait}
+		id = reply[js.replyPrefixLen:]
+		paf = &pubAckFuture{msg: m, jsClient: js.publisher, maxRetries: o.retryAttempts, retryWait: o.retryWait, reply: reply}
 		numPending, maxPending := js.registerPAF(id, paf)
 
 		if maxPending > 0 && numPending > maxPending {
@@ -303,10 +304,17 @@ func (js *jetStream) PublishMsgAsync(m *nats.Msg, opts ...PublishOpt) (PubAckFut
 		}
 	} else {
 		// when retrying, get the ID from existing reply subject
-		id = m.Reply[js.replyPrefixLen:]
+		reply = paf.reply
+		id = reply[js.replyPrefixLen:]
 	}
 
-	if err := js.conn.PublishMsg(m); err != nil {
+	pubMsg := &nats.Msg{
+		Subject: m.Subject,
+		Reply:   reply,
+		Data:    m.Data,
+		Header:  m.Header,
+	}
+	if err := js.conn.PublishMsg(pubMsg); err != nil {
 		js.clearPAF(id)
 		return nil, err
 	}
@@ -370,6 +378,31 @@ func (js *jetStream) handleAsyncReply(m *nats.Msg) {
 		return
 	}
 
+	closeStc := func() {
+		// Check on anyone stalled and waiting.
+		if js.publisher.stallCh != nil && len(js.publisher.acks) < js.publisher.maxpa {
+			close(js.publisher.stallCh)
+			js.publisher.stallCh = nil
+		}
+	}
+
+	closeDchFn := func() func() {
+		var dch chan struct{}
+		// Check on anyone one waiting on done status.
+		if js.publisher.doneCh != nil && len(js.publisher.acks) == 0 {
+			dch = js.publisher.doneCh
+			js.publisher.doneCh = nil
+		}
+		// Return function to close done channel which
+		// should be deferred so that error is processed and
+		// can be checked.
+		return func() {
+			if dch != nil {
+				close(dch)
+			}
+		}
+	}
+
 	doErr := func(err error) {
 		paf.err = err
 		if paf.errCh != nil {
@@ -378,7 +411,6 @@ func (js *jetStream) handleAsyncReply(m *nats.Msg) {
 		cb := js.publisher.asyncPublisherOpts.aecb
 		js.publisher.Unlock()
 		if cb != nil {
-			paf.msg.Reply = ""
 			cb(js, paf.msg, err)
 		}
 	}
@@ -387,7 +419,6 @@ func (js *jetStream) handleAsyncReply(m *nats.Msg) {
 	if len(m.Data) == 0 && m.Header.Get(statusHdr) == noResponders {
 		if paf.retries < paf.maxRetries {
 			paf.retries++
-			paf.msg.Reply = m.Subject
 			time.AfterFunc(paf.retryWait, func() {
 				js.publisher.Lock()
 				paf := js.getPAF(id)
@@ -408,25 +439,16 @@ func (js *jetStream) handleAsyncReply(m *nats.Msg) {
 			return
 		}
 		delete(js.publisher.acks, id)
+		closeStc()
+		defer closeDchFn()()
 		doErr(ErrNoStreamResponse)
 		return
 	}
 
 	// Remove
 	delete(js.publisher.acks, id)
-
-	// Check on anyone stalled and waiting.
-	if js.publisher.stallCh != nil && len(js.publisher.acks) < js.publisher.asyncPublisherOpts.maxpa {
-		close(js.publisher.stallCh)
-		js.publisher.stallCh = nil
-	}
-	// Check on anyone waiting on done status.
-	if js.publisher.doneCh != nil && len(js.publisher.acks) == 0 {
-		dch := js.publisher.doneCh
-		js.publisher.doneCh = nil
-		// Defer here so error is processed and can be checked.
-		defer close(dch)
-	}
+	closeStc()
+	defer closeDchFn()()
 
 	var pa pubAckResponse
 	if err := json.Unmarshal(m.Data, &pa); err != nil {
diff --git a/jetstream/pull.go b/jetstream/pull.go
index 5402828..3869681 100644
--- a/jetstream/pull.go
+++ b/jetstream/pull.go
@@ -971,7 +971,7 @@ func parseMessagesOpts(ordered bool, opts ...PullMessagesOpt) (*consumeOpts, err
 
 func (consumeOpts *consumeOpts) setDefaults(ordered bool) error {
 	if consumeOpts.MaxBytes != unset && consumeOpts.MaxMessages != unset {
-		return fmt.Errorf("only one of MaxMessages and MaxBytes can be specified")
+		return errors.New("only one of MaxMessages and MaxBytes can be specified")
 	}
 	if consumeOpts.MaxBytes != unset {
 		// when max_bytes is used, set batch size to a very large number
@@ -1007,7 +1007,7 @@ func (consumeOpts *consumeOpts) setDefaults(ordered bool) error {
 		}
 	}
 	if consumeOpts.Heartbeat > consumeOpts.Expires/2 {
-		return fmt.Errorf("the value of Heartbeat must be less than 50%% of expiry")
+		return errors.New("the value of Heartbeat must be less than 50%% of expiry")
 	}
 	return nil
 }
diff --git a/jetstream/stream.go b/jetstream/stream.go
index 4741a51..1cd9975 100644
--- a/jetstream/stream.go
+++ b/jetstream/stream.go
@@ -504,16 +504,16 @@ func convertDirectGetMsgResponseToMsg(name string, r *nats.Msg) (*RawStreamMsg,
 	// Check for headers that give us the required information to
 	// reconstruct the message.
 	if len(r.Header) == 0 {
-		return nil, fmt.Errorf("nats: response should have headers")
+		return nil, errors.New("nats: response should have headers")
 	}
 	stream := r.Header.Get(StreamHeader)
 	if stream == "" {
-		return nil, fmt.Errorf("nats: missing stream header")
+		return nil, errors.New("nats: missing stream header")
 	}
 
 	seqStr := r.Header.Get(SequenceHeader)
 	if seqStr == "" {
-		return nil, fmt.Errorf("nats: missing sequence header")
+		return nil, errors.New("nats: missing sequence header")
 	}
 	seq, err := strconv.ParseUint(seqStr, 10, 64)
 	if err != nil {
@@ -521,7 +521,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *nats.Msg) (*RawStreamMsg,
 	}
 	timeStr := r.Header.Get(TimeStampHeaer)
 	if timeStr == "" {
-		return nil, fmt.Errorf("nats: missing timestamp header")
+		return nil, errors.New("nats: missing timestamp header")
 	}
 
 	tm, err := time.Parse(time.RFC3339Nano, timeStr)
@@ -530,7 +530,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *nats.Msg) (*RawStreamMsg,
 	}
 	subj := r.Header.Get(SubjectHeader)
 	if subj == "" {
-		return nil, fmt.Errorf("nats: missing subject header")
+		return nil, errors.New("nats: missing subject header")
 	}
 	return &RawStreamMsg{
 		Subject:  subj,
diff --git a/jetstream/stream_config.go b/jetstream/stream_config.go
index 6eb8432..304203b 100644
--- a/jetstream/stream_config.go
+++ b/jetstream/stream_config.go
@@ -15,6 +15,7 @@ package jetstream
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -584,7 +585,7 @@ func (alg StoreCompression) MarshalJSON() ([]byte, error) {
 	case NoCompression:
 		str = "none"
 	default:
-		return nil, fmt.Errorf("unknown compression algorithm")
+		return nil, errors.New("unknown compression algorithm")
 	}
 	return json.Marshal(str)
 }
@@ -600,7 +601,7 @@ func (alg *StoreCompression) UnmarshalJSON(b []byte) error {
 	case "none":
 		*alg = NoCompression
 	default:
-		return fmt.Errorf("unknown compression algorithm")
+		return errors.New("unknown compression algorithm")
 	}
 	return nil
 }
diff --git a/js.go b/js.go
index 0038afb..e024fae 100644
--- a/js.go
+++ b/js.go
@@ -547,7 +547,7 @@ func (js *js) PublishMsg(m *Msg, opts ...PubOpt) (*PubAck, error) {
 		o.ttl = js.opts.wait
 	}
 	if o.stallWait > 0 {
-		return nil, fmt.Errorf("nats: stall wait cannot be set to sync publish")
+		return nil, errors.New("nats: stall wait cannot be set to sync publish")
 	}
 
 	if o.id != _EMPTY_ {
@@ -1143,7 +1143,7 @@ func RetryAttempts(num int) PubOpt {
 func StallWait(ttl time.Duration) PubOpt {
 	return pubOptFn(func(opts *pubOpts) error {
 		if ttl <= 0 {
-			return fmt.Errorf("nats: stall wait should be more than 0")
+			return errors.New("nats: stall wait should be more than 0")
 		}
 		opts.stallWait = ttl
 		return nil
@@ -1501,11 +1501,11 @@ func processConsInfo(info *ConsumerInfo, userCfg *ConsumerConfig, isPullMode boo
 		// Prevent an user from attempting to create a queue subscription on
 		// a JS consumer that was not created with a deliver group.
 		if queue != _EMPTY_ {
-			return _EMPTY_, fmt.Errorf("cannot create a queue subscription for a consumer without a deliver group")
+			return _EMPTY_, errors.New("cannot create a queue subscription for a consumer without a deliver group")
 		} else if info.PushBound {
 			// Need to reject a non queue subscription to a non queue consumer
 			// if the consumer is already bound.
-			return _EMPTY_, fmt.Errorf("consumer is already bound to a subscription")
+			return _EMPTY_, errors.New("consumer is already bound to a subscription")
 		}
 	} else {
 		// If the JS consumer has a deliver group, we need to fail a non queue
@@ -1607,7 +1607,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 
 	// If no stream name is specified, the subject cannot be empty.
 	if subj == _EMPTY_ && o.stream == _EMPTY_ {
-		return nil, fmt.Errorf("nats: subject required")
+		return nil, errors.New("nats: subject required")
 	}
 
 	// Note that these may change based on the consumer info response we may get.
@@ -1629,7 +1629,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		// would subscribe to and server would send on.
 		if o.cfg.Heartbeat > 0 || o.cfg.FlowControl {
 			// Not making this a public ErrXXX in case we allow in the future.
-			return nil, fmt.Errorf("nats: queue subscription doesn't support idle heartbeat nor flow control")
+			return nil, errors.New("nats: queue subscription doesn't support idle heartbeat nor flow control")
 		}
 
 		// If this is a queue subscription and no consumer nor durable name was specified,
@@ -1667,31 +1667,31 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 	if o.ordered {
 		// Make sure we are not durable.
 		if isDurable {
-			return nil, fmt.Errorf("nats: durable can not be set for an ordered consumer")
+			return nil, errors.New("nats: durable can not be set for an ordered consumer")
 		}
 		// Check ack policy.
 		if o.cfg.AckPolicy != ackPolicyNotSet {
-			return nil, fmt.Errorf("nats: ack policy can not be set for an ordered consumer")
+			return nil, errors.New("nats: ack policy can not be set for an ordered consumer")
 		}
 		// Check max deliver.
 		if o.cfg.MaxDeliver != 1 && o.cfg.MaxDeliver != 0 {
-			return nil, fmt.Errorf("nats: max deliver can not be set for an ordered consumer")
+			return nil, errors.New("nats: max deliver can not be set for an ordered consumer")
 		}
 		// No deliver subject, we pick our own.
 		if o.cfg.DeliverSubject != _EMPTY_ {
-			return nil, fmt.Errorf("nats: deliver subject can not be set for an ordered consumer")
+			return nil, errors.New("nats: deliver subject can not be set for an ordered consumer")
 		}
 		// Queue groups not allowed.
 		if queue != _EMPTY_ {
-			return nil, fmt.Errorf("nats: queues not be set for an ordered consumer")
+			return nil, errors.New("nats: queues not be set for an ordered consumer")
 		}
 		// Check for bound consumers.
 		if consumer != _EMPTY_ {
-			return nil, fmt.Errorf("nats: can not bind existing consumer for an ordered consumer")
+			return nil, errors.New("nats: can not bind existing consumer for an ordered consumer")
 		}
 		// Check for pull mode.
 		if isPullMode {
-			return nil, fmt.Errorf("nats: can not use pull mode for an ordered consumer")
+			return nil, errors.New("nats: can not use pull mode for an ordered consumer")
 		}
 		// Setup how we need it to be here.
 		o.cfg.FlowControl = true
@@ -2425,7 +2425,7 @@ func Description(description string) SubOpt {
 func Durable(consumer string) SubOpt {
 	return subOptFn(func(opts *subOpts) error {
 		if opts.cfg.Durable != _EMPTY_ {
-			return fmt.Errorf("nats: option Durable set more than once")
+			return errors.New("nats: option Durable set more than once")
 		}
 		if opts.consumer != _EMPTY_ && opts.consumer != consumer {
 			return fmt.Errorf("nats: duplicate consumer names (%s and %s)", opts.consumer, consumer)
@@ -3950,7 +3950,7 @@ func (alg StoreCompression) MarshalJSON() ([]byte, error) {
 	case NoCompression:
 		str = "none"
 	default:
-		return nil, fmt.Errorf("unknown compression algorithm")
+		return nil, errors.New("unknown compression algorithm")
 	}
 	return json.Marshal(str)
 }
@@ -3966,7 +3966,7 @@ func (alg *StoreCompression) UnmarshalJSON(b []byte) error {
 	case "none":
 		*alg = NoCompression
 	default:
-		return fmt.Errorf("unknown compression algorithm")
+		return errors.New("unknown compression algorithm")
 	}
 	return nil
 }
diff --git a/jsm.go b/jsm.go
index 6826647..2ae19c7 100644
--- a/jsm.go
+++ b/jsm.go
@@ -1330,11 +1330,11 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 	// Check for headers that give us the required information to
 	// reconstruct the message.
 	if len(r.Header) == 0 {
-		return nil, fmt.Errorf("nats: response should have headers")
+		return nil, errors.New("nats: response should have headers")
 	}
 	stream := r.Header.Get(JSStream)
 	if stream == _EMPTY_ {
-		return nil, fmt.Errorf("nats: missing stream header")
+		return nil, errors.New("nats: missing stream header")
 	}
 
 	// Mirrors can now answer direct gets, so removing check for name equality.
@@ -1342,7 +1342,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 
 	seqStr := r.Header.Get(JSSequence)
 	if seqStr == _EMPTY_ {
-		return nil, fmt.Errorf("nats: missing sequence header")
+		return nil, errors.New("nats: missing sequence header")
 	}
 	seq, err := strconv.ParseUint(seqStr, 10, 64)
 	if err != nil {
@@ -1350,7 +1350,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 	}
 	timeStr := r.Header.Get(JSTimeStamp)
 	if timeStr == _EMPTY_ {
-		return nil, fmt.Errorf("nats: missing timestamp header")
+		return nil, errors.New("nats: missing timestamp header")
 	}
 	// Temporary code: the server in main branch is sending with format
 	// "2006-01-02 15:04:05.999999999 +0000 UTC", but will be changed
@@ -1365,7 +1365,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 	}
 	subj := r.Header.Get(JSSubject)
 	if subj == _EMPTY_ {
-		return nil, fmt.Errorf("nats: missing subject header")
+		return nil, errors.New("nats: missing subject header")
 	}
 	return &RawStreamMsg{
 		Subject:  subj,
diff --git a/kv.go b/kv.go
index 4e7a3fd..3cade1f 100644
--- a/kv.go
+++ b/kv.go
@@ -54,6 +54,7 @@ type KeyValue interface {
 	// Create will add the key/value pair iff it does not exist.
 	Create(key string, value []byte) (revision uint64, err error)
 	// Update will update the value iff the latest revision matches.
+	// Update also resets the TTL associated with the key (if any).
 	Update(key string, value []byte, last uint64) (revision uint64, err error)
 	// Delete will place a delete marker and leave all revisions.
 	Delete(key string, opts ...DeleteOpt) error
diff --git a/micro/README.md b/micro/README.md
index e4ffe09..99949eb 100644
--- a/micro/README.md
+++ b/micro/README.md
@@ -206,35 +206,69 @@ Service IDs can be discovered by:
 ```sh
 nats req '$SRV.PING.EchoService' '' --replies=3
 
-8:59:41 Sending request on "$SRV.PING.EchoService"
-18:59:41 Received with rtt 688.042µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Sending request on "$SRV.PING.EchoService"
+13:03:04 Received with rtt 1.302208ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4K7","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 
-18:59:41 Received with rtt 704.167µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxvO","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Received with rtt 1.317ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4Kt","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 
-18:59:41 Received with rtt 707.875µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdy0a","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Received with rtt 1.320291ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4Lf","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 ```
 
 A specific service instance info can be retrieved:
 
 ```sh
-nats req '$SRV.INFO.EchoService.tNoopzL5Sp1M4qJZdhdxqC' ''
-
-19:40:06 Sending request on "$SRV.INFO.EchoService.tNoopzL5Sp1M4qJZdhdxqC"
-19:40:06 Received with rtt 282.375µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.info_response","description":"","subjects":["svc.echo"]}
+nats req '$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq
+
+13:04:19 Sending request on "$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
+13:04:19 Received with rtt 318.875µs
+{
+  "name": "EchoService",
+  "id": "x3Yuiq7g7MoxhXdxk7i4K7",
+  "version": "1.0.0",
+  "metadata": {},
+  "type": "io.nats.micro.v1.info_response",
+  "description": "",
+  "endpoints": [
+    {
+      "name": "default",
+      "subject": "svc.echo",
+      "queue_group": "q",
+      "metadata": null
+    }
+  ]
+}
 ```
 
 To get statistics for this service:
 
 ```sh
-nats req '$SRV.STATS.EchoService.tNoopzL5Sp1M4qJZdhdxqC' ''
-
-19:40:47 Sending request on "$SRV.STATS.EchoService.tNoopzL5Sp1M4qJZdhdxqC"
-19:40:47 Received with rtt 421.666µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.stats_response","started":"2023-05-22T16:59:39.938514Z","endpoints":[{"name":"default","subject":"svc.echo","metadata":null,"num_requests":0,"num_errors":0,"last_error":"","processing_time":0,"average_processing_time":0}]}
+nats req '$SRV.STATS.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq
+
+13:04:46 Sending request on "$SRV.STATS.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
+13:04:46 Received with rtt 678.25µs
+{
+  "name": "EchoService",
+  "id": "x3Yuiq7g7MoxhXdxk7i4K7",
+  "version": "1.0.0",
+  "metadata": {},
+  "type": "io.nats.micro.v1.stats_response",
+  "started": "2024-09-24T11:02:55.564771Z",
+  "endpoints": [
+    {
+      "name": "default",
+      "subject": "svc.echo",
+      "queue_group": "q",
+      "num_requests": 0,
+      "num_errors": 0,
+      "last_error": "",
+      "processing_time": 0,
+      "average_processing_time": 0
+    }
+  ]
+}
 ```
 
 ## Examples
diff --git a/nats.go b/nats.go
index 10fa0f5..3f12d61 100644
--- a/nats.go
+++ b/nats.go
@@ -86,6 +86,9 @@ const (
 
 	// MAX_CONNECTIONS_ERR is for when nats server denies the connection due to server max_connections limit
 	MAX_CONNECTIONS_ERR = "maximum connections exceeded"
+
+	// MAX_SUBSCRIPTIONS_ERR is for when nats server denies the connection due to server subscriptions limit
+	MAX_SUBSCRIPTIONS_ERR = "maximum subscriptions exceeded"
 )
 
 // Errors
@@ -131,6 +134,7 @@ var (
 	ErrNkeysNotSupported           = errors.New("nats: nkeys not supported by the server")
 	ErrStaleConnection             = errors.New("nats: " + STALE_CONNECTION)
 	ErrTokenAlreadySet             = errors.New("nats: token and token handler both set")
+	ErrUserInfoAlreadySet          = errors.New("nats: cannot set user info callback and user/pass")
 	ErrMsgNotBound                 = errors.New("nats: message is not bound to subscription/connection")
 	ErrMsgNoReply                  = errors.New("nats: message does not have a reply")
 	ErrClientIPNotSupported        = errors.New("nats: client IP not supported by this server")
@@ -140,6 +144,7 @@ var (
 	ErrNoResponders                = errors.New("nats: no responders available for request")
 	ErrMaxConnectionsExceeded      = errors.New("nats: server maximum connections exceeded")
 	ErrConnectionNotTLS            = errors.New("nats: connection is not tls")
+	ErrMaxSubscriptionsExceeded    = errors.New("nats: server maximum subscriptions exceeded")
 )
 
 // GetDefaultOptions returns default configuration options for the client.
@@ -230,6 +235,9 @@ type SignatureHandler func([]byte) ([]byte, error)
 // AuthTokenHandler is used to generate a new token.
 type AuthTokenHandler func() string
 
+// UserInfoCB is used to pass the username and password when establishing connection.
+type UserInfoCB func() (string, string)
+
 // ReconnectDelayHandler is used to get from the user the desired
 // delay the library should pause before attempting to reconnect
 // again. Note that this is invoked after the library tried the
@@ -443,6 +451,9 @@ type Options struct {
 	// Password sets the password to be used when connecting to a server.
 	Password string
 
+	// UserInfo sets the callback handler that will fetch the username and password.
+	UserInfo UserInfoCB
+
 	// Token sets the token to be used when connecting to a server.
 	Token string
 
@@ -1166,6 +1177,13 @@ func UserInfo(user, password string) Option {
 	}
 }
 
+func UserInfoHandler(cb UserInfoCB) Option {
+	return func(o *Options) error {
+		o.UserInfo = cb
+		return nil
+	}
+}
+
 // Token is an Option to set the token to use
 // when a token is not included directly in the URLs
 // and when a token handler is not provided.
@@ -1359,7 +1377,7 @@ func ProxyPath(path string) Option {
 func CustomInboxPrefix(p string) Option {
 	return func(o *Options) error {
 		if p == "" || strings.Contains(p, ">") || strings.Contains(p, "*") || strings.HasSuffix(p, ".") {
-			return fmt.Errorf("nats: invalid custom prefix")
+			return errors.New("nats: invalid custom prefix")
 		}
 		o.InboxPrefix = p
 		return nil
@@ -1814,7 +1832,7 @@ func (nc *Conn) addURLToPool(sURL string, implicit, saveTLSName bool) error {
 	if len(nc.srvPool) == 0 {
 		nc.ws = isWS
 	} else if isWS && !nc.ws || !isWS && nc.ws {
-		return fmt.Errorf("mixing of websocket and non websocket URLs is not allowed")
+		return errors.New("mixing of websocket and non websocket URLs is not allowed")
 	}
 
 	var tlsName string
@@ -2563,6 +2581,13 @@ func (nc *Conn) connectProto() (string, error) {
 		pass = o.Password
 		token = o.Token
 		nkey = o.Nkey
+
+		if nc.Opts.UserInfo != nil {
+			if user != _EMPTY_ || pass != _EMPTY_ {
+				return _EMPTY_, ErrUserInfoAlreadySet
+			}
+			user, pass = nc.Opts.UserInfo()
+		}
 	}
 
 	// Look for user jwt.
@@ -2952,11 +2977,11 @@ func (nc *Conn) doReconnect(err error, forceReconnect bool) {
 
 // processOpErr handles errors from reading or parsing the protocol.
 // The lock should not be held entering this function.
-func (nc *Conn) processOpErr(err error) {
+func (nc *Conn) processOpErr(err error) bool {
 	nc.mu.Lock()
+	defer nc.mu.Unlock()
 	if nc.isConnecting() || nc.isClosed() || nc.isReconnecting() {
-		nc.mu.Unlock()
-		return
+		return false
 	}
 
 	if nc.Opts.AllowReconnect && nc.status == CONNECTED {
@@ -2976,14 +3001,12 @@ func (nc *Conn) processOpErr(err error) {
 		nc.clearPendingFlushCalls()
 
 		go nc.doReconnect(err, false)
-		nc.mu.Unlock()
-		return
+		return false
 	}
 
 	nc.changeConnStatus(DISCONNECTED)
 	nc.err = err
-	nc.mu.Unlock()
-	nc.close(CLOSED, true, nil)
+	return true
 }
 
 // dispatch is responsible for calling any async callbacks
@@ -3080,7 +3103,9 @@ func (nc *Conn) readLoop() {
 			err = nc.parse(buf)
 		}
 		if err != nil {
-			nc.processOpErr(err)
+			if shouldClose := nc.processOpErr(err); shouldClose {
+				nc.close(CLOSED, true, nil)
+			}
 			break
 		}
 	}
@@ -3410,15 +3435,17 @@ slowConsumer:
 	}
 }
 
-// processPermissionsViolation is called when the server signals a subject
-// permissions violation on either publish or subscribe.
-func (nc *Conn) processPermissionsViolation(err string) {
+// processTransientError is called when the server signals a non terminal error
+// which does not close the connection or trigger a reconnect.
+// This will trigger the async error callback if set.
+// These errors include the following:
+// - permissions violation on publish or subscribe
+// - maximum subscriptions exceeded
+func (nc *Conn) processTransientError(err error) {
 	nc.mu.Lock()
-	// create error here so we can pass it as a closure to the async cb dispatcher.
-	e := errors.New("nats: " + err)
-	nc.err = e
+	nc.err = err
 	if nc.Opts.AsyncErrorCB != nil {
-		nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, e) })
+		nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, err) })
 	}
 	nc.mu.Unlock()
 }
@@ -3650,15 +3677,17 @@ func (nc *Conn) processErr(ie string) {
 	// convert to lower case.
 	e := strings.ToLower(ne)
 
-	close := false
+	var close bool
 
 	// FIXME(dlc) - process Slow Consumer signals special.
 	if e == STALE_CONNECTION {
-		nc.processOpErr(ErrStaleConnection)
+		close = nc.processOpErr(ErrStaleConnection)
 	} else if e == MAX_CONNECTIONS_ERR {
-		nc.processOpErr(ErrMaxConnectionsExceeded)
+		close = nc.processOpErr(ErrMaxConnectionsExceeded)
 	} else if strings.HasPrefix(e, PERMISSIONS_ERR) {
-		nc.processPermissionsViolation(ne)
+		nc.processTransientError(fmt.Errorf("nats: %s", ne))
+	} else if strings.HasPrefix(e, MAX_SUBSCRIPTIONS_ERR) {
+		nc.processTransientError(ErrMaxSubscriptionsExceeded)
 	} else if authErr := checkAuthError(e); authErr != nil {
 		nc.mu.Lock()
 		close = nc.processAuthError(authErr)
@@ -5107,7 +5136,9 @@ func (nc *Conn) processPingTimer() {
 	nc.pout++
 	if nc.pout > nc.Opts.MaxPingsOut {
 		nc.mu.Unlock()
-		nc.processOpErr(ErrStaleConnection)
+		if shouldClose := nc.processOpErr(ErrStaleConnection); shouldClose {
+			nc.close(CLOSED, true, nil)
+		}
 		return
 	}
 
@@ -5792,7 +5823,7 @@ func NkeyOptionFromSeed(seedFile string) (Option, error) {
 		return nil, err
 	}
 	if !nkeys.IsValidPublicUserKey(pub) {
-		return nil, fmt.Errorf("nats: Not a valid nkey user seed")
+		return nil, errors.New("nats: Not a valid nkey user seed")
 	}
 	sigCB := func(nonce []byte) ([]byte, error) {
 		return sigHandler(nonce, seedFile)
diff --git a/scripts/cov.sh b/scripts/cov.sh
index 80828cb..fa0fdc1 100755
--- a/scripts/cov.sh
+++ b/scripts/cov.sh
@@ -5,14 +5,15 @@ rm -rf ./cov
 mkdir cov
 go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/nats.out . -tags=skip_no_race_tests
 go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/test.out -coverpkg=github.com/nats-io/nats.go ./test -tags=skip_no_race_tests,internal_testing
-go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/jetstream.out -coverpkg=github.com/nats-io/nats.go/jetstream ./jetstream/test -tags=skip_no_race_tests
+go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/jetstream.out -coverpkg=github.com/nats-io/nats.go/jetstream ./jetstream/...
+go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/service.out -coverpkg=github.com/nats-io/nats.go/micro ./micro/...
 go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/builtin.out -coverpkg=github.com/nats-io/nats.go/encoders/builtin ./test -run EncBuiltin -tags=skip_no_race_tests
 go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/protobuf.out -coverpkg=github.com/nats-io/nats.go/encoders/protobuf ./test -run EncProto -tags=skip_no_race_tests
 gocovmerge ./cov/*.out > acc.out
 rm -rf ./cov
 
 # Without argument, launch browser results. We are going to push to coveralls only
-# from Travis.yml and after success of the build (and result of pushing will not affect
+# from ci.yml and after success of the build (and result of pushing will not affect
 # build result).
 if [[ $1 == "" ]]; then
     go tool cover -html=acc.out
diff --git a/ws.go b/ws.go
index 8cd6c52..fbc5688 100644
--- a/ws.go
+++ b/ws.go
@@ -622,7 +622,7 @@ func (nc *Conn) wsInitHandshake(u *url.URL) error {
 			!strings.EqualFold(resp.Header.Get("Connection"), "upgrade") ||
 			resp.Header.Get("Sec-Websocket-Accept") != wsAcceptKey(wsKey)) {
 
-		err = fmt.Errorf("invalid websocket connection")
+		err = errors.New("invalid websocket connection")
 	}
 	// Check compression extension...
 	if err == nil && compress {
@@ -634,7 +634,7 @@ func (nc *Conn) wsInitHandshake(u *url.URL) error {
 		if !srvCompress {
 			compress = false
 		} else if !noCtxTakeover {
-			err = fmt.Errorf("compression negotiation error")
+			err = errors.New("compression negotiation error")
 		}
 	}
 	if resp != nil {


Test patch:
diff --git a/.github/workflows/latest-server.yaml b/.github/workflows/latest-server.yaml
new file mode 100644
index 0000000..c44523b
--- /dev/null
+++ b/.github/workflows/latest-server.yaml
@@ -0,0 +1,27 @@
+name: Test nats-server@main
+on:
+  schedule:
+    - cron: "30 8 * * *"
+
+jobs:
+    test:
+        runs-on: ubuntu-latest
+        steps:
+            - name: Checkout code
+              uses: actions/checkout@v4
+      
+            - name: Setup Go
+              uses: actions/setup-go@v5
+              with:
+                go-version: 'stable'
+                
+            - name: Get latest server
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go get -modfile go_test.mod github.com/nats-io/nats-server/v2@main
+
+            - name: Test
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
+                go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing
\ No newline at end of file
diff --git a/jetstream/jetstream_test.go b/jetstream/jetstream_test.go
index 58f9064..9d429ad 100644
--- a/jetstream/jetstream_test.go
+++ b/jetstream/jetstream_test.go
@@ -430,13 +430,13 @@ func TestPullConsumer_checkPending(t *testing.T) {
 						}
 						ok <- struct{}{}
 					case <-time.After(1 * time.Second):
-						errs <- fmt.Errorf("Timeout")
+						errs <- errors.New("Timeout")
 						return
 					}
 				} else {
 					select {
 					case <-prChan:
-						errs <- fmt.Errorf("Unexpected pull request")
+						errs <- errors.New("Unexpected pull request")
 					case <-time.After(100 * time.Millisecond):
 						ok <- struct{}{}
 						return
diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go
index 010a7d5..50d2ffa 100644
--- a/jetstream/test/kv_test.go
+++ b/jetstream/test/kv_test.go
@@ -1263,7 +1263,7 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 		checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
 			_, err := kv.Get(context.Background(), key)
 			if err == nil {
-				return fmt.Errorf("Expected key to be gone")
+				return errors.New("Expected key to be gone")
 			}
 			if !errors.Is(err, jetstream.ErrKeyNotFound) {
 				return err
diff --git a/jetstream/test/ordered_test.go b/jetstream/test/ordered_test.go
index 6680955..a6e524b 100644
--- a/jetstream/test/ordered_test.go
+++ b/jetstream/test/ordered_test.go
@@ -1571,21 +1571,19 @@ func TestOrderedConsumerNext(t *testing.T) {
 		}
 
 		publishTestMsgs(t, js)
-		msg, err := c.Next()
+		_, err = c.Next()
 		if err != nil {
 			t.Fatalf("Unexpected error: %s", err)
 		}
-		msg.Ack()
 
 		name := c.CachedInfo().Name
 		if err := s.DeleteConsumer(ctx, name); err != nil {
 			t.Fatal(err)
 		}
-		msg, err = c.Next()
+		_, err = c.Next()
 		if err != nil {
 			t.Fatalf("Unexpected error: %s", err)
 		}
-		msg.Ack()
 	})
 
 	t.Run("consumer used as consume", func(t *testing.T) {
@@ -1623,6 +1621,70 @@ func TestOrderedConsumerNext(t *testing.T) {
 			t.Fatalf("Expected error: %s; got: %s", jetstream.ErrOrderConsumerUsedAsConsume, err)
 		}
 	})
+
+	t.Run("preserve sequence after fetch error", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+			t.Fatalf("Unexpected error during publish: %s", err)
+		}
+		msg, err := c.Next()
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		meta, err := msg.Metadata()
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		if meta.Sequence.Stream != 1 {
+			t.Fatalf("Expected sequence: %d; got: %d", 1, meta.Sequence.Stream)
+		}
+
+		// get next message, it should time out (no more messages on stream)
+		_, err = c.Next(jetstream.FetchMaxWait(100 * time.Millisecond))
+		if !errors.Is(err, nats.ErrTimeout) {
+			t.Fatalf("Expected error: %s; got: %s", nats.ErrTimeout, err)
+		}
+
+		if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+			t.Fatalf("Unexpected error during publish: %s", err)
+		}
+
+		// get next message, it should have stream sequence 2
+		msg, err = c.Next()
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		meta, err = msg.Metadata()
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		if meta.Sequence.Stream != 2 {
+			t.Fatalf("Expected sequence: %d; got: %d", 2, meta.Sequence.Stream)
+		}
+	})
 }
 
 func TestOrderedConsumerFetchNoWait(t *testing.T) {
diff --git a/jetstream/test/publish_test.go b/jetstream/test/publish_test.go
index 2233a16..f79ad19 100644
--- a/jetstream/test/publish_test.go
+++ b/jetstream/test/publish_test.go
@@ -1383,7 +1383,7 @@ func TestPublishAsyncResetPendingOnReconnect(t *testing.T) {
 					errs <- fmt.Errorf("Expected error: %v or %v; got: %v", nats.ErrDisconnected, nats.ErrNoResponders, err)
 				}
 			case <-time.After(5 * time.Second):
-				errs <- fmt.Errorf("Did not receive completion signal")
+				errs <- errors.New("Did not receive completion signal")
 			}
 			wg.Done()
 		}(ack)
@@ -1453,6 +1453,7 @@ func TestPublishAsyncRetry(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
+			publishComplete := js.PublishAsyncComplete()
 			errs := make(chan error, 1)
 			go func() {
 				// create stream with delay so that publish will receive no responders
@@ -1476,6 +1477,12 @@ func TestPublishAsyncRetry(t *testing.T) {
 			case <-time.After(5 * time.Second):
 				t.Fatalf("Timeout waiting for ack")
 			}
+
+			select {
+			case <-publishComplete:
+			case <-time.After(5 * time.Second):
+				t.Fatalf("Did not receive completion signal")
+			}
 		})
 	}
 }
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index b9e0049..1dfafa0 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -504,7 +504,7 @@ func TestAddService(t *testing.T) {
 			}
 
 			if test.givenConfig.ErrorHandler != nil {
-				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.asyncErrorSubject}, fmt.Errorf("oops"))
+				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.asyncErrorSubject}, errors.New("oops"))
 				select {
 				case <-errService:
 				case <-time.After(1 * time.Second):
@@ -536,7 +536,7 @@ func TestAddService(t *testing.T) {
 				}
 			}
 			if test.natsErrorHandler != nil {
-				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.asyncErrorSubject}, fmt.Errorf("oops"))
+				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.asyncErrorSubject}, errors.New("oops"))
 				select {
 				case <-errService:
 					t.Fatalf("Expected to restore nats error handler")
@@ -634,7 +634,7 @@ func TestErrHandlerSubjectMatch(t *testing.T) {
 				}
 				defer svc.Stop()
 
-				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.errSubject}, fmt.Errorf("oops"))
+				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.errSubject}, errors.New("oops"))
 				if test.expectServiceErr {
 					select {
 					case <-errChan:
diff --git a/nats_test.go b/nats_test.go
index e64c801..8f4546a 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -94,10 +94,11 @@ func checkErrChannel(t *testing.T, errCh chan error) {
 }
 
 func TestVersionMatchesTag(t *testing.T) {
-	tag := os.Getenv("TRAVIS_TAG")
-	if tag == "" {
+	refType := os.Getenv("GITHUB_REF_TYPE")
+	if refType != "tag" {
 		t.SkipNow()
 	}
+	tag := os.Getenv("GITHUB_REF_NAME")
 	// We expect a tag of the form vX.Y.Z. If that's not the case,
 	// we need someone to have a look. So fail if first letter is not
 	// a `v`
diff --git a/test/auth_test.go b/test/auth_test.go
index a55b512..8fd0982 100644
--- a/test/auth_test.go
+++ b/test/auth_test.go
@@ -17,6 +17,8 @@ import (
 	"errors"
 	"fmt"
 	"io/fs"
+	"net"
+	"os"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -377,3 +379,75 @@ func TestConnectMissingCreds(t *testing.T) {
 		t.Fatalf("Expected not exists error, got: %v", err)
 	}
 }
+
+func TestUserInfoHandler(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+	listen: 127.0.0.1:-1
+	accounts: {
+	  A {
+		users: [{ user: "pp", password: "foo" }]
+	  }
+	}
+`))
+	defer os.Remove(conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	user, pass := "pp", "foo"
+	userInfoCB := func() (string, string) {
+		return user, pass
+	}
+
+	// check that we cannot set the user info twice
+	_, err := nats.Connect(s.ClientURL(), nats.UserInfo("pp", "foo"), nats.UserInfoHandler(userInfoCB))
+	if !errors.Is(err, nats.ErrUserInfoAlreadySet) {
+		t.Fatalf("Expected ErrUserInfoAlreadySet, got: %v", err)
+	}
+
+	addr, ok := s.Addr().(*net.TCPAddr)
+	if !ok {
+		t.Fatalf("Expected a TCP address, got %T", addr)
+	}
+
+	// check that user/pass from url takes precedence
+	_, err = nats.Connect(fmt.Sprintf("nats://bad:bad@localhost:%d", addr.Port),
+		nats.UserInfoHandler(userInfoCB))
+	if !errors.Is(err, nats.ErrAuthorization) {
+		t.Fatalf("Expected ErrAuthorization, got: %v", err)
+	}
+
+	// connect using the handler
+	nc, err := nats.Connect(s.ClientURL(),
+		nats.ReconnectWait(100*time.Millisecond),
+		nats.UserInfoHandler(userInfoCB))
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
+	}
+	defer nc.Close()
+
+	// now change the password and reload the server
+	newConfig := []byte(`
+	listen: 127.0.0.1:-1
+	accounts: {
+	  A {
+		users: [{ user: "dd", password: "bar" }]
+	  }
+	}
+`)
+	if err := os.WriteFile(conf, newConfig, 0666); err != nil {
+		t.Fatalf("Error writing conf file: %v", err)
+	}
+
+	// update the user info used by the callback
+	user, pass = "dd", "bar"
+
+	status := nc.StatusChanged(nats.CONNECTED)
+
+	if err := s.Reload(); err != nil {
+		t.Fatalf("Error on reload: %v", err)
+	}
+
+	// we should get a reconnected event meaning the new credentials were used
+	WaitOnChannel(t, status, nats.CONNECTED)
+}
diff --git a/test/cluster_test.go b/test/cluster_test.go
index 8501e5d..05fab49 100644
--- a/test/cluster_test.go
+++ b/test/cluster_test.go
@@ -682,7 +682,7 @@ func (d *checkPoolUpdatedDialer) Dial(network, address string) (net.Conn, error)
 		doReal = true
 	} else if d.final {
 		d.ra++
-		return nil, fmt.Errorf("On purpose")
+		return nil, errors.New("On purpose")
 	} else {
 		d.ra++
 		if d.ra == 15 {
@@ -698,7 +698,7 @@ func (d *checkPoolUpdatedDialer) Dial(network, address string) (net.Conn, error)
 		d.conn = c
 		return c, nil
 	}
-	return nil, fmt.Errorf("On purpose")
+	return nil, errors.New("On purpose")
 }
 
 func TestServerPoolUpdatedWhenRouteGoesAway(t *testing.T) {
diff --git a/test/conn_test.go b/test/conn_test.go
index c771355..1a47050 100644
--- a/test/conn_test.go
+++ b/test/conn_test.go
@@ -277,7 +277,7 @@ func TestClientTLSConfig(t *testing.T) {
 		pool := x509.NewCertPool()
 		ok := pool.AppendCertsFromPEM(rootCAs)
 		if !ok {
-			return nil, fmt.Errorf("nats: failed to parse root certificate from")
+			return nil, errors.New("nats: failed to parse root certificate from")
 		}
 		return pool, nil
 	}
@@ -614,7 +614,7 @@ func TestErrOnConnectAndDeadlock(t *testing.T) {
 		nc, err := nats.Connect(natsURL)
 		if err == nil {
 			nc.Close()
-			errCh <- fmt.Errorf("expected bad INFO err, got none")
+			errCh <- errors.New("expected bad INFO err, got none")
 			return
 		}
 		errCh <- nil
@@ -1749,7 +1749,7 @@ type customDialer struct {
 
 func (cd *customDialer) Dial(network, address string) (net.Conn, error) {
 	cd.ch <- true
-	return nil, fmt.Errorf("on purpose")
+	return nil, errors.New("on purpose")
 }
 
 func TestUseCustomDialer(t *testing.T) {
diff --git a/test/drain_test.go b/test/drain_test.go
index c53305e..da07c89 100644
--- a/test/drain_test.go
+++ b/test/drain_test.go
@@ -14,6 +14,7 @@
 package test
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -225,7 +226,7 @@ func TestDrainSlowSubscriber(t *testing.T) {
 		// Wait for it to become invalid. Once drained it is unsubscribed.
 		_, _, err := sub.Pending()
 		if err != nats.ErrBadSubscription {
-			return fmt.Errorf("Still valid")
+			return errors.New("Still valid")
 		}
 		r := int(atomic.LoadInt32(&received))
 		if r != total {
@@ -478,7 +479,7 @@ func TestDrainConnDuringReconnect(t *testing.T) {
 		if nc.IsReconnecting() {
 			return nil
 		}
-		return fmt.Errorf("Not reconnecting yet")
+		return errors.New("Not reconnecting yet")
 	})
 
 	// This should work correctly.
diff --git a/test/js_test.go b/test/js_test.go
index d748a36..db791eb 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -9426,7 +9426,7 @@ func TestJetStreamClusterStreamLeaderChangeClientErr(t *testing.T) {
 						return err
 					}
 					if si.Cluster.Leader == "" {
-						return fmt.Errorf("No leader yet")
+						return errors.New("No leader yet")
 					}
 					return nil
 				})
diff --git a/test/kv_test.go b/test/kv_test.go
index a1b5ca4..7688958 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -1113,7 +1113,7 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 		checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
 			_, err := kv.Get(key)
 			if err == nil {
-				return fmt.Errorf("Expected key to be gone")
+				return errors.New("Expected key to be gone")
 			}
 			if !errors.Is(err, nats.ErrKeyNotFound) {
 				return err
diff --git a/test/sub_test.go b/test/sub_test.go
index 1961e23..559efc5 100644
--- a/test/sub_test.go
+++ b/test/sub_test.go
@@ -16,6 +16,7 @@ package test
 import (
 	"errors"
 	"fmt"
+	"os"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -1319,10 +1320,10 @@ func TestSetPendingLimits(t *testing.T) {
 	// Check for invalid values
 	invalid := func() error {
 		if err := sub.SetPendingLimits(0, 1); err == nil {
-			return fmt.Errorf("Setting limit with 0 should fail")
+			return errors.New("Setting limit with 0 should fail")
 		}
 		if err := sub.SetPendingLimits(1, 0); err == nil {
-			return fmt.Errorf("Setting limit with 0 should fail")
+			return errors.New("Setting limit with 0 should fail")
 		}
 		return nil
 	}
@@ -1737,3 +1738,35 @@ func TestSubscriptionEvents(t *testing.T) {
 		close(blockChan)
 	})
 }
+
+func TestMaxSubscriptionsExceeded(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+	listen: 127.0.0.1:-1
+	max_subscriptions: 5
+	`))
+	defer os.Remove(conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	ch := make(chan error)
+	nc, err := nats.Connect(s.ClientURL(), nats.ErrorHandler(func(c *nats.Conn, s *nats.Subscription, err error) {
+		ch <- err
+	}))
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
+	}
+	defer nc.Close()
+
+	for i := 0; i < 6; i++ {
+		s, err := nc.Subscribe("foo", func(_ *nats.Msg) {})
+		if err != nil {
+			t.Fatalf("Error subscribing: %v", err)
+		}
+		defer s.Unsubscribe()
+	}
+
+	WaitOnChannel(t, ch, nats.ErrMaxSubscriptionsExceeded)
+
+	// wait for the server to process the SUBs
+	time.Sleep(100 * time.Millisecond)
+}
diff --git a/test/ws_test.go b/test/ws_test.go
index 15707d8..ee13b33 100644
--- a/test/ws_test.go
+++ b/test/ws_test.go
@@ -17,6 +17,7 @@ import (
 	"bytes"
 	"crypto/tls"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"math/rand"
 	"net"
@@ -524,7 +525,7 @@ func TestWSStress(t *testing.T) {
 					return
 				}
 				if !bytes.Equal(m.Data[4:4+ps], mainPayload[:ps]) {
-					pushErr(fmt.Errorf("invalid content"))
+					pushErr(errors.New("invalid content"))
 					return
 				}
 				if atomic.AddInt64(&count, 1) == totalRecv {