Repository: nats-io/nats.go
Pull request #1624 (2024-04-27 00:08:26)

[ADDED] Force reconnect
Addresses https://github.com/nats-io/nats-architecture-and-design/issues/259

Signed-off-by: Piotr Piotrowski [piotr@synadia.com](mailto:piotr@synadia.com)

levb: @piotrpio `Reconnect()` ignores `opts.AllowReconnect`, is this by design? (asking for the C client)

Patch:
diff --git a/jetstream/kv.go b/jetstream/kv.go
index 7b39afa..7a026a2 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -197,52 +197,52 @@ type (
 		// Bucket is the name of the KeyValue store. Bucket name has to be
 		// unique and can only contain alphanumeric characters, dashes, and
 		// underscores.
-		Bucket string
+		Bucket string `json:"bucket"`
 
 		// Description is an optional description for the KeyValue store.
-		Description string
+		Description string `json:"description,omitempty"`
 
 		// MaxValueSize is the maximum size of a value in bytes. If not
 		// specified, the default is -1 (unlimited).
-		MaxValueSize int32
+		MaxValueSize int32 `json:"max_value_size,omitempty"`
 
 		// History is the number of historical values to keep per key. If not
 		// specified, the default is 1. Max is 64.
-		History uint8
+		History uint8 `json:"history,omitempty"`
 
 		// TTL is the expiry time for keys. By default, keys do not expire.
-		TTL time.Duration
+		TTL time.Duration `json:"ttl,omitempty"`
 
 		// MaxBytes is the maximum size in bytes of the KeyValue store. If not
 		// specified, the default is -1 (unlimited).
-		MaxBytes int64
+		MaxBytes int64 `json:"max_bytes,omitempty"`
 
 		// Storage is the type of storage to use for the KeyValue store. If not
 		// specified, the default is FileStorage.
-		Storage StorageType
+		Storage StorageType `json:"storage,omitempty"`
 
 		// Replicas is the number of replicas to keep for the KeyValue store in
 		// clustered jetstream. Defaults to 1, maximum is 5.
-		Replicas int
+		Replicas int `json:"num_replicas,omitempty"`
 
 		// Placement is used to declare where the stream should be placed via
 		// tags and/or an explicit cluster name.
-		Placement *Placement
+		Placement *Placement `json:"placement,omitempty"`
 
 		// RePublish allows immediate republishing a message to the configured
 		// subject after it's stored.
-		RePublish *RePublish
+		RePublish *RePublish `json:"republish,omitempty"`
 
 		// Mirror defines the consiguration for mirroring another KeyValue
 		// store.
-		Mirror *StreamSource
+		Mirror *StreamSource `json:"mirror,omitempty"`
 
 		// Sources defines the configuration for sources of a KeyValue store.
-		Sources []*StreamSource
+		Sources []*StreamSource `json:"sources,omitempty"`
 
 		// Compression sets the underlying stream compression.
 		// NOTE: Compression is supported for nats-server 2.10.0+
-		Compression bool
+		Compression bool `json:"compression,omitempty"`
 	}
 
 	// KeyLister is used to retrieve a list of key value store keys. It returns
diff --git a/jsm.go b/jsm.go
index dd09b10..9eb5d4b 100644
--- a/jsm.go
+++ b/jsm.go
@@ -106,51 +106,143 @@ type JetStreamManager interface {
 // There are sensible defaults for most. If no subjects are
 // given the name will be used as the only subject.
 type StreamConfig struct {
-	Name                 string           `json:"name"`
-	Description          string           `json:"description,omitempty"`
-	Subjects             []string         `json:"subjects,omitempty"`
-	Retention            RetentionPolicy  `json:"retention"`
-	MaxConsumers         int              `json:"max_consumers"`
-	MaxMsgs              int64            `json:"max_msgs"`
-	MaxBytes             int64            `json:"max_bytes"`
-	Discard              DiscardPolicy    `json:"discard"`
-	DiscardNewPerSubject bool             `json:"discard_new_per_subject,omitempty"`
-	MaxAge               time.Duration    `json:"max_age"`
-	MaxMsgsPerSubject    int64            `json:"max_msgs_per_subject"`
-	MaxMsgSize           int32            `json:"max_msg_size,omitempty"`
-	Storage              StorageType      `json:"storage"`
-	Replicas             int              `json:"num_replicas"`
-	NoAck                bool             `json:"no_ack,omitempty"`
-	Template             string           `json:"template_owner,omitempty"`
-	Duplicates           time.Duration    `json:"duplicate_window,omitempty"`
-	Placement            *Placement       `json:"placement,omitempty"`
-	Mirror               *StreamSource    `json:"mirror,omitempty"`
-	Sources              []*StreamSource  `json:"sources,omitempty"`
-	Sealed               bool             `json:"sealed,omitempty"`
-	DenyDelete           bool             `json:"deny_delete,omitempty"`
-	DenyPurge            bool             `json:"deny_purge,omitempty"`
-	AllowRollup          bool             `json:"allow_rollup_hdrs,omitempty"`
-	Compression          StoreCompression `json:"compression"`
-	FirstSeq             uint64           `json:"first_seq,omitempty"`
-
-	// Allow applying a subject transform to incoming messages before doing anything else.
+	// Name is the name of the stream. It is required and must be unique
+	// across the JetStream account.
+	//
+	// Name Names cannot contain whitespace, ., *, >, path separators
+	// (forward or backwards slash), and non-printable characters.
+	Name string `json:"name"`
+
+	// Description is an optional description of the stream.
+	Description string `json:"description,omitempty"`
+
+	// Subjects is a list of subjects that the stream is listening on.
+	// Wildcards are supported. Subjects cannot be set if the stream is
+	// created as a mirror.
+	Subjects []string `json:"subjects,omitempty"`
+
+	// Retention defines the message retention policy for the stream.
+	// Defaults to LimitsPolicy.
+	Retention RetentionPolicy `json:"retention"`
+
+	// MaxConsumers specifies the maximum number of consumers allowed for
+	// the stream.
+	MaxConsumers int `json:"max_consumers"`
+
+	// MaxMsgs is the maximum number of messages the stream will store.
+	// After reaching the limit, stream adheres to the discard policy.
+	// If not set, server default is -1 (unlimited).
+	MaxMsgs int64 `json:"max_msgs"`
+
+	// MaxBytes is the maximum total size of messages the stream will store.
+	// After reaching the limit, stream adheres to the discard policy.
+	// If not set, server default is -1 (unlimited).
+	MaxBytes int64 `json:"max_bytes"`
+
+	// Discard defines the policy for handling messages when the stream
+	// reaches its limits in terms of number of messages or total bytes.
+	Discard DiscardPolicy `json:"discard"`
+
+	// DiscardNewPerSubject is a flag to enable discarding new messages per
+	// subject when limits are reached. Requires DiscardPolicy to be
+	// DiscardNew and the MaxMsgsPerSubject to be set.
+	DiscardNewPerSubject bool `json:"discard_new_per_subject,omitempty"`
+
+	// MaxAge is the maximum age of messages that the stream will retain.
+	MaxAge time.Duration `json:"max_age"`
+
+	// MaxMsgsPerSubject is the maximum number of messages per subject that
+	// the stream will retain.
+	MaxMsgsPerSubject int64 `json:"max_msgs_per_subject"`
+
+	// MaxMsgSize is the maximum size of any single message in the stream.
+	MaxMsgSize int32 `json:"max_msg_size,omitempty"`
+
+	// Storage specifies the type of storage backend used for the stream
+	// (file or memory).
+	Storage StorageType `json:"storage"`
+
+	// Replicas is the number of stream replicas in clustered JetStream.
+	// Defaults to 1, maximum is 5.
+	Replicas int `json:"num_replicas"`
+
+	// NoAck is a flag to disable acknowledging messages received by this
+	// stream.
+	//
+	// If set to true, publish methods from the JetStream client will not
+	// work as expected, since they rely on acknowledgements. Core NATS
+	// publish methods should be used instead. Note that this will make
+	// message delivery less reliable.
+	NoAck bool `json:"no_ack,omitempty"`
+
+	// Duplicates is the window within which to track duplicate messages.
+	// If not set, server default is 2 minutes.
+	Duplicates time.Duration `json:"duplicate_window,omitempty"`
+
+	// Placement is used to declare where the stream should be placed via
+	// tags and/or an explicit cluster name.
+	Placement *Placement `json:"placement,omitempty"`
+
+	// Mirror defines the configuration for mirroring another stream.
+	Mirror *StreamSource `json:"mirror,omitempty"`
+
+	// Sources is a list of other streams this stream sources messages from.
+	Sources []*StreamSource `json:"sources,omitempty"`
+
+	// Sealed streams do not allow messages to be published or deleted via limits or API,
+	// sealed streams can not be unsealed via configuration update. Can only
+	// be set on already created streams via the Update API.
+	Sealed bool `json:"sealed,omitempty"`
+
+	// DenyDelete restricts the ability to delete messages from a stream via
+	// the API. Defaults to false.
+	DenyDelete bool `json:"deny_delete,omitempty"`
+
+	// DenyPurge restricts the ability to purge messages from a stream via
+	// the API. Defaults to false.
+	DenyPurge bool `json:"deny_purge,omitempty"`
+
+	// AllowRollup allows the use of the Nats-Rollup header to replace all
+	// contents of a stream, or subject in a stream, with a single new
+	// message.
+	AllowRollup bool `json:"allow_rollup_hdrs,omitempty"`
+
+	// Compression specifies the message storage compression algorithm.
+	// Defaults to NoCompression.
+	Compression StoreCompression `json:"compression"`
+
+	// FirstSeq is the initial sequence number of the first message in the
+	// stream.
+	FirstSeq uint64 `json:"first_seq,omitempty"`
+
+	// SubjectTransform allows applying a transformation to matching
+	// messages' subjects.
 	SubjectTransform *SubjectTransformConfig `json:"subject_transform,omitempty"`
 
-	// Allow republish of the message after being sequenced and stored.
+	// RePublish allows immediate republishing a message to the configured
+	// subject after it's stored.
 	RePublish *RePublish `json:"republish,omitempty"`
 
-	// Allow higher performance, direct access to get individual messages. E.g. KeyValue
+	// AllowDirect enables direct access to individual messages using direct
+	// get API. Defaults to false.
 	AllowDirect bool `json:"allow_direct"`
-	// Allow higher performance and unified direct access for mirrors as well.
+
+	// MirrorDirect enables direct access to individual messages from the
+	// origin stream using direct get API. Defaults to false.
 	MirrorDirect bool `json:"mirror_direct"`
 
-	// Limits for consumers on this stream.
+	// ConsumerLimits defines limits of certain values that consumers can
+	// set, defaults for those who don't set these settings
 	ConsumerLimits StreamConsumerLimits `json:"consumer_limits,omitempty"`
 
-	// Metadata is additional metadata for the Stream.
-	// Keys starting with `_nats` are reserved.
-	// NOTE: Metadata requires nats-server v2.10.0+
+	// Metadata is a set of application-defined key-value pairs for
+	// associating metadata on the stream. This feature requires nats-server
+	// v2.10.0 or later.
 	Metadata map[string]string `json:"metadata,omitempty"`
+
+	// Template identifies the template that manages the Stream. DEPRECATED:
+	// This feature is no longer supported.
+	Template string `json:"template_owner,omitempty"`
 }
 
 // SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received.
diff --git a/kv.go b/kv.go
index 8ed84be..d9f40fd 100644
--- a/kv.go
+++ b/kv.go
@@ -249,22 +249,22 @@ func purge() DeleteOpt {
 
 // KeyValueConfig is for configuring a KeyValue store.
 type KeyValueConfig struct {
-	Bucket       string
-	Description  string
-	MaxValueSize int32
-	History      uint8
-	TTL          time.Duration
-	MaxBytes     int64
-	Storage      StorageType
-	Replicas     int
-	Placement    *Placement
-	RePublish    *RePublish
-	Mirror       *StreamSource
-	Sources      []*StreamSource
+	Bucket       string          `json:"bucket"`
+	Description  string          `json:"description,omitempty"`
+	MaxValueSize int32           `json:"max_value_size,omitempty"`
+	History      uint8           `json:"history,omitempty"`
+	TTL          time.Duration   `json:"ttl,omitempty"`
+	MaxBytes     int64           `json:"max_bytes,omitempty"`
+	Storage      StorageType     `json:"storage,omitempty"`
+	Replicas     int             `json:"num_replicas,omitempty"`
+	Placement    *Placement      `json:"placement,omitempty"`
+	RePublish    *RePublish      `json:"republish,omitempty"`
+	Mirror       *StreamSource   `json:"mirror,omitempty"`
+	Sources      []*StreamSource `json:"sources,omitempty"`
 
 	// Enable underlying stream compression.
 	// NOTE: Compression is supported for nats-server 2.10.0+
-	Compression bool
+	Compression bool `json:"compression,omitempty"`
 }
 
 // Used to watch all keys.
diff --git a/nats.go b/nats.go
index 8c0796a..d94c9a9 100644
--- a/nats.go
+++ b/nats.go
@@ -2161,6 +2161,47 @@ func (nc *Conn) waitForExits() {
 	nc.wg.Wait()
 }
 
+// ForceReconnect forces a reconnect attempt to the server.
+// This is a non-blocking call and will start the reconnect
+// process without waiting for it to complete.
+//
+// If the connection is already in the process of reconnecting,
+// this call will force an immediate reconnect attempt (bypassing
+// the current reconnect delay).
+func (nc *Conn) ForceReconnect() error {
+	nc.mu.Lock()
+	defer nc.mu.Unlock()
+
+	if nc.isClosed() {
+		return ErrConnectionClosed
+	}
+	if nc.isReconnecting() {
+		// if we're already reconnecting, force a reconnect attempt
+		// even if we're in the middle of a backoff
+		if nc.rqch != nil {
+			close(nc.rqch)
+		}
+		return nil
+	}
+
+	// Clear any queued pongs
+	nc.clearPendingFlushCalls()
+
+	// Clear any queued and blocking requests.
+	nc.clearPendingRequestCalls()
+
+	// Stop ping timer if set.
+	nc.stopPingTimer()
+
+	// Go ahead and make sure we have flushed the outbound
+	nc.bw.flush()
+	nc.conn.Close()
+
+	nc.changeConnStatus(RECONNECTING)
+	go nc.doReconnect(nil, true)
+	return nil
+}
+
 // ConnectedUrl reports the connected server's URL
 func (nc *Conn) ConnectedUrl() string {
 	if nc == nil {
@@ -2420,7 +2461,7 @@ func (nc *Conn) connect() (bool, error) {
 		nc.setup()
 		nc.changeConnStatus(RECONNECTING)
 		nc.bw.switchToPending()
-		go nc.doReconnect(ErrNoServers)
+		go nc.doReconnect(ErrNoServers, false)
 		err = nil
 	} else {
 		nc.current = nil
@@ -2720,7 +2761,7 @@ func (nc *Conn) stopPingTimer() {
 
 // Try to reconnect using the option parameters.
 // This function assumes we are allowed to reconnect.
-func (nc *Conn) doReconnect(err error) {
+func (nc *Conn) doReconnect(err error, forceReconnect bool) {
 	// We want to make sure we have the other watchers shutdown properly
 	// here before we proceed past this point.
 	nc.waitForExits()
@@ -2776,7 +2817,8 @@ func (nc *Conn) doReconnect(err error) {
 			break
 		}
 
-		doSleep := i+1 >= len(nc.srvPool)
+		doSleep := i+1 >= len(nc.srvPool) && !forceReconnect
+		forceReconnect = false
 		nc.mu.Unlock()
 
 		if !doSleep {
@@ -2803,6 +2845,12 @@ func (nc *Conn) doReconnect(err error) {
 			select {
 			case <-rqch:
 				rt.Stop()
+
+				// we need to reset the rqch channel to avoid
+				// closing a closed channel in the next iteration
+				nc.mu.Lock()
+				nc.rqch = make(chan struct{})
+				nc.mu.Unlock()
 			case <-rt.C:
 			}
 		}
@@ -2872,9 +2920,6 @@ func (nc *Conn) doReconnect(err error) {
 		// Done with the pending buffer
 		nc.bw.doneWithPending()
 
-		// This is where we are truly connected.
-		nc.status = CONNECTED
-
 		// Queue up the correct callback. If we are in initial connect state
 		// (using retry on failed connect), we will call the ConnectedCB,
 		// otherwise the ReconnectedCB.
@@ -2930,7 +2975,7 @@ func (nc *Conn) processOpErr(err error) {
 		// Clear any queued pongs, e.g. pending flush calls.
 		nc.clearPendingFlushCalls()
 
-		go nc.doReconnect(err)
+		go nc.doReconnect(err, false)
 		nc.mu.Unlock()
 		return
 	}


Test patch:
diff --git a/example_test.go b/example_test.go
index 6aa9363..3ad3673 100644
--- a/example_test.go
+++ b/example_test.go
@@ -89,6 +89,19 @@ func ExampleConn_Subscribe() {
 	})
 }
 
+func ExampleConn_ForceReconnect() {
+	nc, _ := nats.Connect(nats.DefaultURL)
+	defer nc.Close()
+
+	nc.Subscribe("foo", func(m *nats.Msg) {
+		fmt.Printf("Received a message: %s\n", string(m.Data))
+	})
+
+	// Reconnect to the server.
+	// the subscription will be recreated after the reconnect.
+	nc.ForceReconnect()
+}
+
 // This Example shows a synchronous subscriber.
 func ExampleConn_SubscribeSync() {
 	nc, _ := nats.Connect(nats.DefaultURL)
diff --git a/test/conn_test.go b/test/conn_test.go
index 7e5fcab..afc5025 100644
--- a/test/conn_test.go
+++ b/test/conn_test.go
@@ -2946,16 +2946,6 @@ func TestRetryOnFailedConnectWithTLSError(t *testing.T) {
 }
 
 func TestConnStatusChangedEvents(t *testing.T) {
-	waitForStatus := func(t *testing.T, ch chan nats.Status, expected nats.Status) {
-		select {
-		case s := <-ch:
-			if s != expected {
-				t.Fatalf("Expected status: %s; got: %s", expected, s)
-			}
-		case <-time.After(5 * time.Second):
-			t.Fatalf("Timeout waiting for status %q", expected)
-		}
-	}
 	t.Run("default events", func(t *testing.T) {
 		s := RunDefaultServer()
 		nc, err := nats.Connect(s.ClientURL())
@@ -2978,15 +2968,15 @@ func TestConnStatusChangedEvents(t *testing.T) {
 		time.Sleep(50 * time.Millisecond)
 
 		s.Shutdown()
-		waitForStatus(t, newStatus, nats.RECONNECTING)
+		WaitOnChannel(t, newStatus, nats.RECONNECTING)
 
 		s = RunDefaultServer()
 		defer s.Shutdown()
 
-		waitForStatus(t, newStatus, nats.CONNECTED)
+		WaitOnChannel(t, newStatus, nats.CONNECTED)
 
 		nc.Close()
-		waitForStatus(t, newStatus, nats.CLOSED)
+		WaitOnChannel(t, newStatus, nats.CLOSED)
 
 		select {
 		case s := <-newStatus:
@@ -3019,7 +3009,7 @@ func TestConnStatusChangedEvents(t *testing.T) {
 		s = RunDefaultServer()
 		defer s.Shutdown()
 		nc.Close()
-		waitForStatus(t, newStatus, nats.CLOSED)
+		WaitOnChannel(t, newStatus, nats.CLOSED)
 
 		select {
 		case s := <-newStatus:
diff --git a/test/helper_test.go b/test/helper_test.go
index 9c04a40..7f2aedf 100644
--- a/test/helper_test.go
+++ b/test/helper_test.go
@@ -54,6 +54,18 @@ func WaitTime(ch chan bool, timeout time.Duration) error {
 	return errors.New("timeout")
 }
 
+func WaitOnChannel[T comparable](t *testing.T, ch <-chan T, expected T) {
+	t.Helper()
+	select {
+	case s := <-ch:
+		if s != expected {
+			t.Fatalf("Expected result: %v; got: %v", expected, s)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Timeout waiting for result %v", expected)
+	}
+}
+
 func stackFatalf(t tLogger, f string, args ...any) {
 	lines := make([]string, 0, 32)
 	msg := fmt.Sprintf(f, args...)
diff --git a/test/reconnect_test.go b/test/reconnect_test.go
index 66cc9b6..e543db7 100644
--- a/test/reconnect_test.go
+++ b/test/reconnect_test.go
@@ -853,7 +853,7 @@ func TestAuthExpiredReconnect(t *testing.T) {
 
 	jwtCB := func() (string, error) {
 		claims := jwt.NewUserClaims("test")
-		claims.Expires = time.Now().Add(500 * time.Millisecond).Unix()
+		claims.Expires = time.Now().Add(time.Second).Unix()
 		claims.Subject = upub
 		jwt, err := claims.Encode(akp)
 		if err != nil {
@@ -884,21 +884,218 @@ func TestAuthExpiredReconnect(t *testing.T) {
 	case <-time.After(2 * time.Second):
 		t.Fatal("Did not get the auth expired error")
 	}
-	select {
-	case s := <-stasusCh:
-		if s != nats.RECONNECTING {
-			t.Fatalf("Expected to be in reconnecting state after jwt expires, got %v", s)
+	WaitOnChannel(t, stasusCh, nats.RECONNECTING)
+	WaitOnChannel(t, stasusCh, nats.CONNECTED)
+	nc.Close()
+}
+
+func TestForceReconnect(t *testing.T) {
+	s := RunDefaultServer()
+
+	nc, err := nats.Connect(s.ClientURL(), nats.ReconnectWait(10*time.Second))
+	if err != nil {
+		t.Fatalf("Unexpected error on connect: %v", err)
+	}
+
+	statusCh := nc.StatusChanged(nats.RECONNECTING, nats.CONNECTED)
+	defer close(statusCh)
+	newStatus := make(chan nats.Status, 10)
+	// non-blocking channel, so we need to be constantly listening
+	go func() {
+		for {
+			s, ok := <-statusCh
+			if !ok {
+				return
+			}
+			newStatus <- s
 		}
-	case <-time.After(2 * time.Second):
-		t.Fatal("Did not get the status change")
+	}()
+
+	sub, err := nc.SubscribeSync("foo")
+	if err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+	if err := nc.Publish("foo", []byte("msg")); err != nil {
+		t.Fatalf("Error on publish: %v", err)
+	}
+	_, err = sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+
+	// Force a reconnect
+	err = nc.ForceReconnect()
+	if err != nil {
+		t.Fatalf("Unexpected error on reconnect: %v", err)
+	}
+
+	WaitOnChannel(t, newStatus, nats.RECONNECTING)
+	WaitOnChannel(t, newStatus, nats.CONNECTED)
+
+	if err := nc.Publish("foo", []byte("msg")); err != nil {
+		t.Fatalf("Error on publish: %v", err)
+	}
+	_, err = sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+
+	// shutdown server and then force a reconnect
+	s.Shutdown()
+	WaitOnChannel(t, newStatus, nats.RECONNECTING)
+	_, err = sub.NextMsg(100 * time.Millisecond)
+	if err == nil {
+		t.Fatal("Expected error getting message")
+	}
+
+	// restart server
+	s = RunDefaultServer()
+	defer s.Shutdown()
+
+	if err := nc.ForceReconnect(); err != nil {
+		t.Fatalf("Unexpected error on reconnect: %v", err)
+	}
+	// wait for the reconnect
+	// because the connection has long ReconnectWait,
+	// if force reconnect does not work, the test will timeout
+	WaitOnChannel(t, newStatus, nats.CONNECTED)
+
+	if err := nc.Publish("foo", []byte("msg")); err != nil {
+		t.Fatalf("Error on publish: %v", err)
+	}
+	_, err = sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+	nc.Close()
+}
+
+func TestForceReconnectDisallowReconnect(t *testing.T) {
+	s := RunDefaultServer()
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL(), nats.NoReconnect())
+	if err != nil {
+		t.Fatalf("Unexpected error on connect: %v", err)
+	}
+	defer nc.Close()
+
+	statusCh := nc.StatusChanged(nats.RECONNECTING, nats.CONNECTED)
+	defer close(statusCh)
+	newStatus := make(chan nats.Status, 10)
+	// non-blocking channel, so we need to be constantly listening
+	go func() {
+		for {
+			s, ok := <-statusCh
+			if !ok {
+				return
+			}
+			newStatus <- s
+		}
+	}()
+
+	sub, err := nc.SubscribeSync("foo")
+	if err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+	if err := nc.Publish("foo", []byte("msg")); err != nil {
+		t.Fatalf("Error on publish: %v", err)
+	}
+	_, err = sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+
+	// Force a reconnect
+	err = nc.ForceReconnect()
+	if err != nil {
+		t.Fatalf("Unexpected error on reconnect: %v", err)
 	}
+
+	WaitOnChannel(t, newStatus, nats.RECONNECTING)
+	WaitOnChannel(t, newStatus, nats.CONNECTED)
+
+	if err := nc.Publish("foo", []byte("msg")); err != nil {
+		t.Fatalf("Error on publish: %v", err)
+	}
+	_, err = sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+
+}
+
+func TestAuthExpiredForceReconnect(t *testing.T) {
+	ts := runTrustServer()
+	defer ts.Shutdown()
+
+	_, err := nats.Connect(ts.ClientURL())
+	if err == nil {
+		t.Fatalf("Expecting an error on connect")
+	}
+	ukp, err := nkeys.FromSeed(uSeed)
+	if err != nil {
+		t.Fatalf("Error creating user key pair: %v", err)
+	}
+	upub, err := ukp.PublicKey()
+	if err != nil {
+		t.Fatalf("Error getting user public key: %v", err)
+	}
+	akp, err := nkeys.FromSeed(aSeed)
+	if err != nil {
+		t.Fatalf("Error creating account key pair: %v", err)
+	}
+
+	jwtCB := func() (string, error) {
+		claims := jwt.NewUserClaims("test")
+		claims.Expires = time.Now().Add(time.Second).Unix()
+		claims.Subject = upub
+		jwt, err := claims.Encode(akp)
+		if err != nil {
+			return "", err
+		}
+		return jwt, nil
+	}
+	sigCB := func(nonce []byte) ([]byte, error) {
+		kp, _ := nkeys.FromSeed(uSeed)
+		sig, _ := kp.Sign(nonce)
+		return sig, nil
+	}
+
+	errCh := make(chan error, 1)
+	nc, err := nats.Connect(ts.ClientURL(), nats.UserJWT(jwtCB, sigCB), nats.ReconnectWait(10*time.Second),
+		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+			errCh <- err
+		}))
+	if err != nil {
+		t.Fatalf("Expected to connect, got %v", err)
+	}
+	defer nc.Close()
+	statusCh := nc.StatusChanged(nats.RECONNECTING, nats.CONNECTED)
+	defer close(statusCh)
+	newStatus := make(chan nats.Status, 10)
+	// non-blocking channel, so we need to be constantly listening
+	go func() {
+		for {
+			s, ok := <-statusCh
+			if !ok {
+				return
+			}
+			newStatus <- s
+		}
+	}()
+	time.Sleep(100 * time.Millisecond)
 	select {
-	case s := <-stasusCh:
-		if s != nats.CONNECTED {
-			t.Fatalf("Expected to reconnect, got %v", s)
+	case err := <-errCh:
+		if !errors.Is(err, nats.ErrAuthExpired) {
+			t.Fatalf("Expected auth expired error, got %v", err)
 		}
 	case <-time.After(2 * time.Second):
-		t.Fatal("Did not get the status change")
+		t.Fatal("Did not get the auth expired error")
 	}
-	nc.Close()
+	if err := nc.ForceReconnect(); err != nil {
+		t.Fatalf("Unexpected error on reconnect: %v", err)
+	}
+	WaitOnChannel(t, newStatus, nats.RECONNECTING)
+	WaitOnChannel(t, newStatus, nats.CONNECTED)
 }
diff --git a/test/sub_test.go b/test/sub_test.go
index 0bf2880..f0f83a8 100644
--- a/test/sub_test.go
+++ b/test/sub_test.go
@@ -1617,18 +1617,6 @@ func TestSubscribe_ClosedHandler(t *testing.T) {
 }
 
 func TestSubscriptionEvents(t *testing.T) {
-
-	waitForStatus := func(t *testing.T, ch <-chan nats.SubStatus, expected nats.SubStatus) {
-		t.Helper()
-		select {
-		case s := <-ch:
-			if s != expected {
-				t.Fatalf("Expected status: %s; got: %s", expected, s)
-			}
-		case <-time.After(5 * time.Second):
-			t.Fatalf("Timeout waiting for status %q", expected)
-		}
-	}
 	t.Run("default events", func(t *testing.T) {
 		s := RunDefaultServer()
 		defer s.Shutdown()
@@ -1651,19 +1639,19 @@ func TestSubscriptionEvents(t *testing.T) {
 		status := sub.StatusChanged()
 
 		// initial status
-		waitForStatus(t, status, nats.SubscriptionActive)
+		WaitOnChannel(t, status, nats.SubscriptionActive)
 
 		for i := 0; i < 11; i++ {
 			nc.Publish("foo", []byte("Hello"))
 		}
-		waitForStatus(t, status, nats.SubscriptionSlowConsumer)
+		WaitOnChannel(t, status, nats.SubscriptionSlowConsumer)
 		close(blockChan)
 
 		sub.Drain()
 
-		waitForStatus(t, status, nats.SubscriptionDraining)
+		WaitOnChannel(t, status, nats.SubscriptionDraining)
 
-		waitForStatus(t, status, nats.SubscriptionClosed)
+		WaitOnChannel(t, status, nats.SubscriptionClosed)
 	})
 
 	t.Run("slow consumer event only", func(t *testing.T) {
@@ -1691,7 +1679,7 @@ func TestSubscriptionEvents(t *testing.T) {
 		for i := 0; i < 20; i++ {
 			nc.Publish("foo", []byte("Hello"))
 		}
-		waitForStatus(t, status, nats.SubscriptionSlowConsumer)
+		WaitOnChannel(t, status, nats.SubscriptionSlowConsumer)
 		close(blockChan)
 
 		// now try with sync sub
@@ -1706,7 +1694,7 @@ func TestSubscriptionEvents(t *testing.T) {
 		for i := 0; i < 20; i++ {
 			nc.Publish("foo", []byte("Hello"))
 		}
-		waitForStatus(t, status, nats.SubscriptionSlowConsumer)
+		WaitOnChannel(t, status, nats.SubscriptionSlowConsumer)
 	})
 
 	t.Run("do not block channel if it's not read", func(t *testing.T) {
@@ -1730,7 +1718,7 @@ func TestSubscriptionEvents(t *testing.T) {
 		}
 		sub.SetPendingLimits(10, 1024)
 		status := sub.StatusChanged()
-		waitForStatus(t, status, nats.SubscriptionActive)
+		WaitOnChannel(t, status, nats.SubscriptionActive)
 
 		// chan length is 10, so make sure we switch state more times
 		for i := 0; i < 20; i++ {