Repository: nats-io/nats.go (go)

Issue:
Explicit support for auth change during reconnect
### Proposed change

I am experiencing "fuzzy" behavior regarding the following. Fuzzy as in, the code works when the NATS client is connected to a single URL, but it doesnt work when connected to multiple URLs. 

We are using the NATS Auth Callout, but client side, the Password field needs to change each time the disconnect callback is called, which happens when the auth callout JWTs expire (eg hourly). Specifically, our autth callout backend is interpreting a "password" specific to our in house IAM stack, and those *also* expire. 

Because our auth callout tokens expire every hour, our `disconnectedCB` is called every hour. And in that, we want to change `conn.Opts.Password` so that it uses the new password (the client has to regenerate one at this time) on the reconnection attempt (see below, the actual auth call is made after the reconnected callback, so "hook wise", we already have a hook at the right spot.)

My functions currently looks like this:

```go
func (gb *Base) tokenAndConnect(ctx context.Context) (*nats.Conn, error) {
	iamCreds, err := gb.iamRuntime.GetAccessToken(ctx, &identity.GetAccessTokenRequest{})
	if err != nil {
		gb.Logger.Fatalf("failed to get access token: %v", err)
	}

	// For the auth callout, we have a sentinel and we also need an Identity token as our password
	return nats.Connect(
		gb.config.Nats.URL,
		nats.UserCredentials(gb.config.Nats.SentinelCredsFile),
		nats.UserInfo(gb.config.MsName, iamCreds.Token),
	)
}


func (gb *Base) AddNATSConn(ctx context.Context) *Base {
	nc, err := gb.tokenAndConnect(ctx)
	if err != nil {
		gb.Logger.Fatalf("failed to create Initial NATS connection: %v", err)
	}

	/*
		    Disconnect Callbacks are called when the function reconnects....
			https://github.com/nats-io/nats.go/blob/370bc4fc4cac2de8d5dbbddbb98739f9cb2d7f8b/nats.go#L2779-L2787
			The purpose of this is to get a new identityJWT before the reconnect happens

			We cant use the reconnectCB (misleading name here)
			ReconnectCB is called *after* the reconnection happens. In order to refresh the token before the reconnect,
			Doc strings for variouis callbacks:
			ReconnectedCB sets the reconnected handler called whenever the connection is successfully reconnected.
			ConnectedCB sets the connected handler called when the initial connection is established. It is not invoked on successful reconnects - for reconnections, use ReconnectedCB
			DisconnectedErrCB sets the disconnected error handler that is called whenever the connection is disconnected.
			ClosedCB sets the closed handler that is called when a client will no longer be connected.
	*/

	nc.SetDisconnectErrHandler(func(conn *nats.Conn, err error) {
		gb.Logger.Infof("NATS connection disconnected: %v. Generating new Identity JWT", err)
		iamCreds, err := gb.iamRuntime.GetAccessToken(ctx, &identity.GetAccessTokenRequest{})
		if err != nil {
			gb.Logger.Fatalf("Failed to get access token: %w", err)
		}
		
                 // *********************************!!!
                 // doesnt work in multi URL case but works fine when you seed the client with a single NATS server
		conn.Opts.Password = iamCreds.Token
	})

	nc.SetReconnectHandler(func(conn *nats.Conn) {
		gb.Logger.Infof("NATS connection reconnected. Reconnect count %d", conn.Reconnects)
	})

	gb.NatsConn = nc
	return gb
}
```


I think it has something to do with this warning which I just saw..
https://github.com/nats-io/nats.go/blob/e3df53d7fa82d1d60335c53b66b43d9be90a8881/nats.go#L537-L538

It works in the case where the client only was given one cluster URL but not multiple. 

I eventually tracked down this call chain on reconnects:
1. the disconnect handler is called first: https://github.com/nats-io/nats.go/blob/e3df53d7fa82d1d60335c53b66b43d9be90a8881/nats.go#L2781-L2787
2. later in that function, createConn is called though I dont think this is the problematic one: https://github.com/nats-io/nats.go/blob/e3df53d7fa82d1d60335c53b66b43d9be90a8881/nats.go#L2873 
3. then later down in that function, nc.processConnectInit() is called: https://github.com/nats-io/nats.go/blob/e3df53d7fa82d1d60335c53b66b43d9be90a8881/nats.go#L2886
4. that calls sendConnect(): https://github.com/nats-io/nats.go/blob/e3df53d7fa82d1d60335c53b66b43d9be90a8881/nats.go#L2380
5. that calls connectProto https://github.com/nats-io/nats.go/blob/e3df53d7fa82d1d60335c53b66b43d9be90a8881/nats.go#L2644
6. that contains: https://github.com/nats-io/nats.go/blob/e3df53d7fa82d1d60335c53b66b43d9be90a8881/nats.go#L2551-L2566

that code makes it unclear when this `Password` change is going to be reflected. 

My fear/suspicion is that, in the multiple URL case, that `User` is populated; triggering the `if`, which is why it sends the old token, which is why it breaks. In the single URL case, that is not populated, thus triggering the `else`

I am trying to see how I can explicitly force  it to use a new password, either via setting some variable sure to be picked up, or via a callback. Maybe a flag that says "always override the auth from the connection options". 

This block might be related, I was trying to find anywhere where `nc.current.url` is set: https://github.com/nats-io/nats.go/blob/e3df53d7fa82d1d60335c53b66b43d9be90a8881/nats.go#L1821-L1827

### Use case

We are using the NATS Auth Callout, but client side, the Password field needs to change each time the disconnect callback is called, which happens when the auth callout JWTs expire (eg hourly). Specifically, our autth callout backend is interpreting a "password" specific to our in house IAM stack.

### Contribution

_No response_

Pull request description:
[ADDED] UserInfoHandler for dynamically setting user/password
Resolves #1694

Signed-off-by: Piotr Piotrowski <piotr@synadia.com>

Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 4a37564..6b3b32d 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -35,7 +35,7 @@ jobs:
 
         strategy:
             matrix:
-              go: [ "1.21", "1.22" ]
+              go: [ "1.22", "1.23" ]
         steps:
             - name: Checkout code
               uses: actions/checkout@v4
@@ -55,14 +55,14 @@ jobs:
               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.22" ]; then
+                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.22'
+              if: matrix.go == '1.23'
               uses: coverallsapp/github-action@v2
               with:
                file: acc.out
\ No newline at end of file
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index e401cb9..36a44c8 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -435,7 +435,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/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..0d7f952 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -747,7 +747,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 +774,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/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/nats.go b/nats.go
index 10fa0f5..83638a5 100644
--- a/nats.go
+++ b/nats.go
@@ -131,6 +131,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")
@@ -230,6 +231,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 +447,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 +1173,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 +1373,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 +1828,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 +2577,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.
@@ -5792,7 +5813,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/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/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/publish_test.go b/jetstream/test/publish_test.go
index 2233a16..2ab6de8 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)
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/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..a27e142 100644
--- a/test/sub_test.go
+++ b/test/sub_test.go
@@ -1319,10 +1319,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
 	}
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 {