Repository: nats-io/nats-server
Pull request #3441 (2022-09-06 15:44:20)

Do not auto-promote direct get functionality. 
 Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core


ripienaar: I dont thnk, for this, we need the client update - `omitempty` is a noop in the context of this PR. So can safely revert, however this is perhaps due to the consumer name/durable_name change?
kozlovic: @ripienaar And some other changes to error handling where the library returns now a tiny different error text (but also returns full fledge JS API error where one can inspect the code). So that will require quite a bit of careful update of tests. So if that PR can be done without client update, maybe that's the way to go and have a pass after 2.9.0?
ripienaar: Thanks @kozlovic  been going through and saw that as well, surprised the client library landed that in main thought it was part of the new simplification erfforts only.

I dont think the update is needed - however this is a worrying indication of compatibility issues this close to 2.9.
ripienaar: I tested just removing these 3 lines that force the change without any omitempty updates, client updates etc - and using old clients that has omitempty.  

The issue remains resolved without client updates

Patch:
diff --git a/go.mod b/go.mod
index 4141c02d..a21285e5 100644
--- a/go.mod
+++ b/go.mod
@@ -1,18 +1,18 @@
 module github.com/nats-io/nats-server/v2
 
-go 1.18
+go 1.19
 
 require (
 	github.com/klauspost/compress v1.15.9
 	github.com/minio/highwayhash v1.0.2
 	github.com/nats-io/jwt/v2 v2.3.0
-	github.com/nats-io/nats.go v1.16.1-0.20220810192301-fb5ca2cbc995
+	github.com/nats-io/nats.go v1.16.1-0.20220906180156-a1017eec10b0
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.5.1
-	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e
-	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a
-	golang.org/x/time v0.0.0-20220411224347-583f2d630306
+	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90
+	golang.org/x/sys v0.0.0-20220906135438-9e1f76180b77
+	golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9
 )
 
 require (
diff --git a/go.sum b/go.sum
index 22ff41fd..33cb8c7b 100644
--- a/go.sum
+++ b/go.sum
@@ -15,8 +15,8 @@ github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
 github.com/nats-io/jwt/v2 v2.3.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
-github.com/nats-io/nats.go v1.16.1-0.20220810192301-fb5ca2cbc995 h1:CUcSQR8jwa9//qNgN/t3tW53DObnTPQ/G/K+qnS7yRc=
-github.com/nats-io/nats.go v1.16.1-0.20220810192301-fb5ca2cbc995/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats.go v1.16.1-0.20220906180156-a1017eec10b0 h1:dPUKD6Iv8M1y9MU8PK6H4a4/12yx5/CbaYWz/Z1arY8=
+github.com/nats-io/nats.go v1.16.1-0.20220906180156-a1017eec10b0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -27,17 +27,17 @@ github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMT
 go.uber.org/automaxprocs v1.5.1 h1:e1YG66Lrk73dn4qhg8WFSvhF0JuFQF0ERIp4rpuV8Qk=
 go.uber.org/automaxprocs v1.5.1/go.mod h1:BF4eumQw0P9GtnuxxovUd06vwm1o18oMzFtK66vU6XU=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e h1:T8NU3HyQ8ClP4SEE+KbFlg6n0NhuTsN4MyznaarGsZM=
-golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
+golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a h1:dGzPydgVsqGcTRVwiLJ1jVbufYwmzD3LfVPLKsKg+0k=
-golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220906135438-9e1f76180b77 h1:C1tElbkWrsSkn3IRl1GCW/gETw1TywWIPgwZtXTZbYg=
+golang.org/x/sys v0.0.0-20220906135438-9e1f76180b77/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/time v0.0.0-20220411224347-583f2d630306 h1:+gHMid33q6pen7kv9xvT+JRinntgeXO2AeZVd0AWD3w=
-golang.org/x/time v0.0.0-20220411224347-583f2d630306/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9 h1:ftMN5LMiBFjbzleLqtoBZk7KdJwhuybIU+FckUHgoyQ=
+golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
diff --git a/server/stream.go b/server/stream.go
index 41c85ac7..20c47ec1 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -61,9 +61,9 @@ type StreamConfig struct {
 	RePublish *RePublish `json:"republish,omitempty"`
 
 	// Allow higher performance, direct access to get individual messages. E.g. KeyValue
-	AllowDirect bool `json:"allow_direct,omitempty"`
+	AllowDirect bool `json:"allow_direct"`
 	// Allow higher performance and unified direct access for mirrors as well.
-	MirrorDirect bool `json:"mirror_direct,omitempty"`
+	MirrorDirect bool `json:"mirror_direct"`
 
 	// Optional qualifiers. These can not be modified after set to true.
 
@@ -1246,11 +1246,6 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		}
 	}
 
-	// Here we will auto set direct gets if MaxMsgsPerSubject is set.
-	if cfg.MaxMsgsPer > 0 {
-		cfg.AllowDirect = true
-	}
-
 	return cfg, nil
 }
 
diff --git a/server/sublist.go b/server/sublist.go
index ef0b283b..bb204ad5 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -427,6 +427,7 @@ func (s *Sublist) Insert(sub *subscription) error {
 	} else {
 		if n.qsubs == nil {
 			n.qsubs = make(map[string]map[*subscription]*subscription)
+			isnew = true
 		}
 		qname := string(sub.queue)
 		// This is a queue subscription
@@ -434,7 +435,6 @@ func (s *Sublist) Insert(sub *subscription) error {
 		if !ok {
 			subs = make(map[*subscription]*subscription)
 			n.qsubs[qname] = subs
-			isnew = true
 		}
 		subs[sub] = sub
 	}
@@ -892,10 +892,8 @@ func (s *Sublist) removeFromNode(n *node, sub *subscription) (found, last bool)
 	_, found = qsub[sub]
 	delete(qsub, sub)
 	if len(qsub) == 0 {
-		// This is the last queue subscription interest when len(qsub) == 0, not
-		// when n.qsubs is empty.
-		last = true
 		delete(n.qsubs, string(sub.queue))
+		last = len(n.qsubs) == 0
 	}
 	return found, last
 }


Test patch:
diff --git a/server/benchmark_publish_test.go b/server/benchmark_publish_test.go
deleted file mode 100644
index c10b2979..00000000
--- a/server/benchmark_publish_test.go
+++ /dev/null
@@ -1,186 +0,0 @@
-// Copyright 2022 The NATS Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package server
-
-import (
-	"fmt"
-	"math/rand"
-	"sync/atomic"
-	"testing"
-
-	"github.com/nats-io/nats.go"
-)
-
-func BenchmarkPublish(b *testing.B) {
-
-	const (
-		verbose     = false
-		seed        = 12345
-		minMessages = 10_000
-		subject     = "S"
-		queue       = "Q"
-	)
-
-	const (
-		KB = 1024
-		MB = KB * KB
-	)
-
-	type SubscriberType string
-	const (
-		Async      SubscriberType = "Async"
-		QueueAsync SubscriberType = "AsyncQueue"
-		None       SubscriberType = "None"
-	)
-
-	benchmarksCases := []struct {
-		messageSize int
-	}{
-		{0},
-		{1},
-		{32},
-		{128},
-		{512},
-		{4 * KB},
-		{32 * KB},
-		{128 * KB},
-		{512 * KB},
-		{1 * MB},
-	}
-
-	// All the cases above are run for each of the subscriber cases below
-	subscribersCases := []struct {
-		numSubs int
-		subType SubscriberType
-	}{
-		{0, None},
-		{1, Async},
-		{1, QueueAsync},
-		{10, Async},
-		{10, QueueAsync},
-	}
-
-	for _, bc := range benchmarksCases {
-		bcName := fmt.Sprintf(
-			"MsgSz=%db",
-			bc.messageSize,
-		)
-
-		b.Run(
-			bcName,
-			func(b *testing.B) {
-
-				for _, sc := range subscribersCases {
-
-					scName := fmt.Sprintf("Subs=%dx%v", sc.numSubs, sc.subType)
-					if sc.subType == None {
-						scName = fmt.Sprintf("Subs=%v", sc.subType)
-					}
-
-					b.Run(
-						scName,
-						func(b *testing.B) {
-							// Skip short runs, benchmark gets re-executed with a larger N
-							if b.N < minMessages {
-								b.ResetTimer()
-								return
-							}
-
-							if verbose {
-								b.Logf("Running %s/%s with %d ops", bcName, scName, b.N)
-							}
-
-							subErrors := uint64(0)
-							handleSubError := func(_ *nats.Conn, _ *nats.Subscription, _ error) {
-								atomic.AddUint64(&subErrors, 1)
-							}
-
-							// Start single server (no JS)
-							opts := DefaultTestOptions
-							opts.Port = -1
-							s := RunServer(&opts)
-							defer s.Shutdown()
-
-							// Create subscribers
-							for i := 0; i < sc.numSubs; i++ {
-								subConn, connErr := nats.Connect(s.ClientURL(), nats.ErrorHandler(handleSubError))
-								if connErr != nil {
-									b.Fatalf("Failed to connect: %v", connErr)
-								}
-								defer subConn.Close()
-
-								var sub *nats.Subscription
-								var subErr error
-
-								switch sc.subType {
-								case None:
-									// No subscription
-								case Async:
-									sub, subErr = subConn.Subscribe(subject, func(*nats.Msg) {})
-								case QueueAsync:
-									sub, subErr = subConn.QueueSubscribe(subject, queue, func(*nats.Msg) {})
-								default:
-									b.Fatalf("Unknow subscribers type: %v", sc.subType)
-								}
-
-								if subErr != nil {
-									b.Fatalf("Failed to subscribe: %v", subErr)
-								}
-								defer sub.Unsubscribe()
-								// Do not drop messages due to slow subscribers:
-								sub.SetPendingLimits(-1, -1)
-							}
-
-							// Create publisher connection
-							nc, err := nats.Connect(s.ClientURL())
-							if err != nil {
-								b.Fatalf("Failed to connect: %v", err)
-							}
-							defer nc.Close()
-
-							rng := rand.New(rand.NewSource(int64(seed)))
-							message := make([]byte, bc.messageSize)
-							var published, errors int
-
-							// Benchmark starts here
-							b.ResetTimer()
-
-							for i := 0; i < b.N; i++ {
-								rng.Read(message)
-								pubErr := nc.Publish(subject, message)
-								if pubErr != nil {
-									errors++
-								} else {
-									published++
-									b.SetBytes(int64(bc.messageSize))
-								}
-							}
-
-							// Benchmark ends here
-							b.StopTimer()
-
-							if published+errors != b.N {
-								b.Fatalf("Something doesn't add up: %d + %d != %d", published, errors, b.N)
-							} else if subErrors > 0 {
-								b.Fatalf("Subscribers errors: %d", subErrors)
-							}
-
-							b.ReportMetric(float64(errors)*100/float64(b.N), "%error")
-						},
-					)
-				}
-			},
-		)
-	}
-}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 93ab7216..e5f7efcc 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -1458,10 +1458,8 @@ func TestJetStreamClusterStreamExtendedUpdates(t *testing.T) {
 	// Mirror changes are not supported for now
 	cfg.Subjects = nil
 	cfg.Mirror = &nats.StreamSource{Name: "ORDERS"}
-	if _, err := js.UpdateStream(cfg); err == nil ||
-		!strings.Contains(NewJSStreamMirrorNotUpdatableError().Error(), err.Error()) {
-		t.Fatalf("Expected error %q, got %q", NewJSStreamMirrorNotUpdatableError(), err)
-	}
+	_, err := js.UpdateStream(cfg)
+	require_Error(t, err, NewJSStreamMirrorNotUpdatableError())
 }
 
 func TestJetStreamClusterDoubleAdd(t *testing.T) {
@@ -5665,7 +5663,7 @@ func TestJetStreamClusterSourceFilterSubjectUpdateFail(t *testing.T) {
 		Replicas: 2,
 	})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "source 'TEST' filter subject 'notthere' does not overlap with any origin stream subject")
+	require_Equal(t, err.Error(), "nats: source 'TEST' filter subject 'notthere' does not overlap with any origin stream subject")
 
 	_, err = js.AddStream(&nats.StreamConfig{
 		Name:     "M",
@@ -5673,7 +5671,7 @@ func TestJetStreamClusterSourceFilterSubjectUpdateFail(t *testing.T) {
 		Replicas: 2,
 	})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "mirror 'TEST' filter subject 'notthere' does not overlap with any origin stream subject")
+	require_Equal(t, err.Error(), "nats: mirror 'TEST' filter subject 'notthere' does not overlap with any origin stream subject")
 }
 
 func TestJetStreamClusterMirrorAndSourcesFilteredConsumers(t *testing.T) {
@@ -9313,9 +9311,8 @@ func TestJetStreamClusterAccountInfoForSystemAccount(t *testing.T) {
 	nc, js := jsClientConnect(t, c.randomServer(), nats.UserInfo("admin", "s3cr3t!"))
 	defer nc.Close()
 
-	if _, err := js.AccountInfo(); err != nats.ErrJetStreamNotEnabled {
-		t.Fatalf("Expected a not enabled error for system account, got %v", err)
-	}
+	_, err := js.AccountInfo()
+	require_Error(t, err, nats.ErrJetStreamNotEnabledForAccount)
 }
 
 func TestJetStreamClusterListFilter(t *testing.T) {
@@ -9387,6 +9384,7 @@ func TestJetStreamClusterConsumerUpdates(t *testing.T) {
 	testConsumerUpdate := func(t *testing.T, s *Server, replicas int) {
 		nc, js := jsClientConnect(t, s)
 		defer nc.Close()
+
 		// Create a stream.
 		_, err := js.AddStream(&nats.StreamConfig{
 			Name:     "TEST",
@@ -9409,7 +9407,6 @@ func TestJetStreamClusterConsumerUpdates(t *testing.T) {
 			MaxDeliver:     5,
 			MaxAckPending:  50,
 		}
-
 		_, err = js.AddConsumer("TEST", cfg)
 		require_NoError(t, err)
 
@@ -9432,13 +9429,13 @@ func TestJetStreamClusterConsumerUpdates(t *testing.T) {
 
 		// Description
 		cfg.Description = "New Description"
-		_, err = js.AddConsumer("TEST", cfg)
+		_, err = js.UpdateConsumer("TEST", cfg)
 		require_NoError(t, err)
 
 		// MaxAckPending
 		checkSubsPending(t, sub, 50)
 		cfg.MaxAckPending = 75
-		_, err = js.AddConsumer("TEST", cfg)
+		_, err = js.UpdateConsumer("TEST", cfg)
 		require_NoError(t, err)
 		checkSubsPending(t, sub, 75)
 
@@ -9454,54 +9451,54 @@ func TestJetStreamClusterConsumerUpdates(t *testing.T) {
 		// AckWait
 		checkSubsPending(t, sub, 0)
 		cfg.AckWait = 200 * time.Millisecond
-		_, err = js.AddConsumer("TEST", cfg)
+		_, err = js.UpdateConsumer("TEST", cfg)
 		require_NoError(t, err)
 		checkSubsPending(t, sub, 10)
 
 		// Rate Limit
 		cfg.RateLimit = 8 * 1024
-		_, err = js.AddConsumer("TEST", cfg)
+		_, err = js.UpdateConsumer("TEST", cfg)
 		require_NoError(t, err)
 
 		cfg.RateLimit = 0
-		_, err = js.AddConsumer("TEST", cfg)
+		_, err = js.UpdateConsumer("TEST", cfg)
 		require_NoError(t, err)
 
 		// These all should fail.
 		ncfg = *cfg
 		ncfg.DeliverPolicy = nats.DeliverLastPolicy
-		_, err = js.AddConsumer("TEST", &ncfg)
+		_, err = js.UpdateConsumer("TEST", &ncfg)
 		require_Error(t, err)
 
 		ncfg = *cfg
 		ncfg.OptStartSeq = 22
-		_, err = js.AddConsumer("TEST", &ncfg)
+		_, err = js.UpdateConsumer("TEST", &ncfg)
 		require_Error(t, err)
 
 		ncfg = *cfg
 		now := time.Now()
 		ncfg.OptStartTime = &now
-		_, err = js.AddConsumer("TEST", &ncfg)
+		_, err = js.UpdateConsumer("TEST", &ncfg)
 		require_Error(t, err)
 
 		ncfg = *cfg
 		ncfg.AckPolicy = nats.AckAllPolicy
-		_, err = js.AddConsumer("TEST", &ncfg)
+		_, err = js.UpdateConsumer("TEST", &ncfg)
 		require_Error(t, err)
 
 		ncfg = *cfg
 		ncfg.ReplayPolicy = nats.ReplayOriginalPolicy
-		_, err = js.AddConsumer("TEST", &ncfg)
+		_, err = js.UpdateConsumer("TEST", &ncfg)
 		require_Error(t, err)
 
 		ncfg = *cfg
 		ncfg.Heartbeat = time.Second
-		_, err = js.AddConsumer("TEST", &ncfg)
+		_, err = js.UpdateConsumer("TEST", &ncfg)
 		require_Error(t, err)
 
 		ncfg = *cfg
 		ncfg.FlowControl = true
-		_, err = js.AddConsumer("TEST", &ncfg)
+		_, err = js.UpdateConsumer("TEST", &ncfg)
 		require_Error(t, err)
 
 	}
@@ -9581,7 +9578,7 @@ func TestJetStreamClusterAccountReservations(t *testing.T) {
 
 		_, err = js.AddStream(&nats.StreamConfig{Name: "S2", Subjects: []string{"s2"}, MaxBytes: 1024, Replicas: replica})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "insufficient storage resources available")
+		require_Equal(t, err.Error(), "nats: insufficient storage resources available")
 
 		_, err = js.UpdateStream(&nats.StreamConfig{Name: "S1", Subjects: []string{"s1"}, MaxBytes: mb / 2, Replicas: replica})
 		require_NoError(t, err)
@@ -9591,11 +9588,11 @@ func TestJetStreamClusterAccountReservations(t *testing.T) {
 
 		_, err = js.AddStream(&nats.StreamConfig{Name: "S3", Subjects: []string{"s3"}, MaxBytes: 1024, Replicas: replica})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "insufficient storage resources available")
+		require_Equal(t, err.Error(), "nats: insufficient storage resources available")
 
 		_, err = js.UpdateStream(&nats.StreamConfig{Name: "S2", Subjects: []string{"s2"}, MaxBytes: mb/2 + 1, Replicas: replica})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "insufficient storage resources available")
+		require_Equal(t, err.Error(), "nats: insufficient storage resources available")
 
 		require_NoError(t, js.DeleteStream("S1"))
 		require_NoError(t, js.DeleteStream("S2"))
@@ -9631,7 +9628,7 @@ func TestJetStreamClusterConcurrentAccountLimits(t *testing.T) {
 		})
 		if err != nil {
 			atomic.AddInt32(&failCount, 1)
-			require_Equal(t, err.Error(), "insufficient storage resources available")
+			require_Equal(t, err.Error(), "nats: insufficient storage resources available")
 		}
 	}
 
@@ -11243,7 +11240,7 @@ func TestJetStreamClusterMirrorSourceLoop(t *testing.T) {
 			Sources:  []*nats.StreamSource{{Name: "1"}},
 		})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "detected cycle")
+		require_Equal(t, err.Error(), "nats: detected cycle")
 	}
 
 	t.Run("Single", func(t *testing.T) {
diff --git a/server/jetstream_jwt_test.go b/server/jetstream_jwt_test.go
index 916418fd..935cba20 100644
--- a/server/jetstream_jwt_test.go
+++ b/server/jetstream_jwt_test.go
@@ -468,10 +468,10 @@ func TestJetStreamJWTClusteredTiers(t *testing.T) {
 	// Test absent tiers
 	_, err = js.AddStream(&nats.StreamConfig{Name: "testR2", Replicas: 2, Subjects: []string{"testR2"}})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "no JetStream default or applicable tiered limit present")
+	require_Equal(t, err.Error(), "nats: no JetStream default or applicable tiered limit present")
 	_, err = js.AddStream(&nats.StreamConfig{Name: "testR5", Replicas: 5, Subjects: []string{"testR5"}})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "no JetStream default or applicable tiered limit present")
+	require_Equal(t, err.Error(), "nats: no JetStream default or applicable tiered limit present")
 
 	// Test tiers up to stream limits
 	_, err = js.AddStream(&nats.StreamConfig{Name: "testR1-1", Replicas: 1, Subjects: []string{"testR1-1"}})
@@ -484,10 +484,10 @@ func TestJetStreamJWTClusteredTiers(t *testing.T) {
 	// Test exceeding tiered stream limit
 	_, err = js.AddStream(&nats.StreamConfig{Name: "testR1-3", Replicas: 1, Subjects: []string{"testR1-3"}})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum number of streams reached")
+	require_Equal(t, err.Error(), "nats: maximum number of streams reached")
 	_, err = js.AddStream(&nats.StreamConfig{Name: "testR3-3", Replicas: 3, Subjects: []string{"testR3-3"}})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum number of streams reached")
+	require_Equal(t, err.Error(), "nats: maximum number of streams reached")
 
 	// Test tiers up to consumer limits
 	_, err = js.AddConsumer("testR1-1", &nats.ConsumerConfig{Durable: "dur1", AckPolicy: nats.AckExplicitPolicy})
@@ -500,10 +500,10 @@ func TestJetStreamJWTClusteredTiers(t *testing.T) {
 	// test exceeding tiered consumer limits
 	_, err = js.AddConsumer("testR1-1", &nats.ConsumerConfig{Durable: "dur4", AckPolicy: nats.AckExplicitPolicy})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum consumers limit reached")
+	require_Equal(t, err.Error(), "nats: maximum consumers limit reached")
 	_, err = js.AddConsumer("testR1-1", &nats.ConsumerConfig{Durable: "dur5", AckPolicy: nats.AckExplicitPolicy})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum consumers limit reached")
+	require_Equal(t, err.Error(), "nats: maximum consumers limit reached")
 
 	// test tiered storage limit
 	msg := [512]byte{}
@@ -633,7 +633,7 @@ func TestJetStreamJWTClusteredTiersChange(t *testing.T) {
 	cfg.Replicas = 3
 	_, err = js.UpdateStream(cfg)
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "insufficient storage resources available")
+	require_Equal(t, err.Error(), "nats: insufficient storage resources available")
 
 	time.Sleep(time.Second - time.Since(start)) // make sure the time stamp changes
 	accClaim.Limits.JetStreamTieredLimits["R3"] = jwt.JetStreamLimits{
diff --git a/server/jetstream_super_cluster_test.go b/server/jetstream_super_cluster_test.go
index 6e18c499..668d0e38 100644
--- a/server/jetstream_super_cluster_test.go
+++ b/server/jetstream_super_cluster_test.go
@@ -1387,7 +1387,7 @@ func TestJetStreamSuperClusterOverflowPlacement(t *testing.T) {
 		MaxBytes:  2 * 1024 * 1024 * 1024,
 		Placement: &nats.Placement{Cluster: pcn},
 	})
-	require_Contains(t, err.Error(), "no suitable peers for placement")
+	require_Contains(t, err.Error(), "nats: no suitable peers for placement")
 	// Now test actual overflow placement. So try again with no placement designation.
 	// This will test the peer picker's logic since they are updated at this point and the meta leader
 	// knows it can not place it in C2.
@@ -2391,7 +2391,7 @@ func TestJetStreamSuperClusterMaxHaAssets(t *testing.T) {
 	waitStatsz(3, 1)
 	_, err = js.AddStream(&nats.StreamConfig{Name: "S3", Replicas: 3, Placement: &nats.Placement{Cluster: "C1"}})
 	require_Error(t, err)
-	require_Contains(t, err.Error(), "no suitable peers for placement")
+	require_Contains(t, err.Error(), "nats: no suitable peers for placement")
 	require_Contains(t, err.Error(), "misc: 3")
 	require_NoError(t, js.DeleteStream("S1"))
 	waitStatsz(3, 2)
@@ -2402,7 +2402,7 @@ func TestJetStreamSuperClusterMaxHaAssets(t *testing.T) {
 	waitStatsz(3, 1)
 	_, err = js.AddConsumer("S2", &nats.ConsumerConfig{Durable: "DUR2", AckPolicy: nats.AckExplicitPolicy})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "insufficient resources")
+	require_Equal(t, err.Error(), "nats: insufficient resources")
 	_, err = js.AddConsumer("S2", &nats.ConsumerConfig{AckPolicy: nats.AckExplicitPolicy})
 	require_NoError(t, err)
 	waitStatsz(3, 3)
@@ -2422,10 +2422,10 @@ func TestJetStreamSuperClusterMaxHaAssets(t *testing.T) {
 	waitStatsz(6, 3)
 	_, err = js.AddConsumer("S4", &nats.ConsumerConfig{Durable: "DUR2", AckPolicy: nats.AckExplicitPolicy})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "insufficient resources")
+	require_Equal(t, err.Error(), "nats: insufficient resources")
 	_, err = js.UpdateStream(&nats.StreamConfig{Name: "S2", Replicas: 3, Placement: &nats.Placement{Cluster: "C2"}})
 	require_Error(t, err)
-	require_Contains(t, err.Error(), "no suitable peers for placement")
+	require_Contains(t, err.Error(), "nats: no suitable peers for placement")
 	require_Contains(t, err.Error(), "misc: 3")
 }
 
@@ -3294,6 +3294,7 @@ func TestJetStreamSuperClusterMirrorInheritsAllowDirect(t *testing.T) {
 		Subjects:          []string{"key.*"},
 		Placement:         &nats.Placement{Tags: []string{"cloud:aws", "country:us"}},
 		MaxMsgsPerSubject: 1,
+		AllowDirect:       true,
 	})
 	require_NoError(t, err)
 
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 900d222b..d02adf0e 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -11933,7 +11933,7 @@ func TestJetStreamMirrorUpdatePreventsSubjects(t *testing.T) {
 	require_NoError(t, err)
 
 	_, err = js.UpdateStream(&nats.StreamConfig{Name: "MIRROR", Mirror: &nats.StreamSource{Name: "ORIGINAL"}, Subjects: []string{"x"}})
-	if err == nil || err.Error() != "stream mirrors can not contain subjects" {
+	if err == nil || err.Error() != "nats: stream mirrors can not contain subjects" {
 		t.Fatalf("Expected to not be able to put subjects on a stream, got: %+v", err)
 	}
 }
@@ -13584,7 +13584,7 @@ func TestJetStreamNegativeDupeWindow(t *testing.T) {
 		Replicas:          1,
 		NoAck:             false,
 	})
-	if err == nil || err.Error() != "duplicates window can not be negative" {
+	if err == nil || err.Error() != "nats: duplicates window can not be negative" {
 		t.Fatalf("Expected dupe window error got: %v", err)
 	}
 }
@@ -14974,12 +14974,20 @@ func TestJetStreamConsumerUpdateSurvival(t *testing.T) {
 	_, err := js.AddStream(&nats.StreamConfig{Name: "X"})
 	require_NoError(t, err)
 
-	// First create a consumer that is push based.
-	_, err = js.AddConsumer("X", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy, MaxAckPending: 1024})
+	// First create a consumer with max ack pending.
+	_, err = js.AddConsumer("X", &nats.ConsumerConfig{
+		Durable:       "dlc",
+		AckPolicy:     nats.AckExplicitPolicy,
+		MaxAckPending: 1024,
+	})
 	require_NoError(t, err)
 
 	// Now do same name but pull. This will update the MaxAcKPending
-	ci, err := js.AddConsumer("X", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy, MaxAckPending: 22})
+	ci, err := js.UpdateConsumer("X", &nats.ConsumerConfig{
+		Durable:       "dlc",
+		AckPolicy:     nats.AckExplicitPolicy,
+		MaxAckPending: 22,
+	})
 	require_NoError(t, err)
 
 	if ci.Config.MaxAckPending != 22 {
@@ -16703,11 +16711,11 @@ func TestJetStreamLimits(t *testing.T) {
 
 		_, err = js.UpdateStream(&nats.StreamConfig{Name: "bar", Duplicates: 2 * time.Minute})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "duplicates window can not be larger then server limit of 1m0s")
+		require_Equal(t, err.Error(), "nats: duplicates window can not be larger then server limit of 1m0s")
 
 		_, err = js.AddStream(&nats.StreamConfig{Name: "baz", Duplicates: 2 * time.Minute})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "duplicates window can not be larger then server limit of 1m0s")
+		require_Equal(t, err.Error(), "nats: duplicates window can not be larger then server limit of 1m0s")
 
 		ci, err := js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "dur1", AckPolicy: nats.AckExplicitPolicy})
 		require_NoError(t, err)
@@ -16716,7 +16724,7 @@ func TestJetStreamLimits(t *testing.T) {
 
 		_, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "dur2", AckPolicy: nats.AckExplicitPolicy, MaxRequestBatch: 500})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "consumer max request batch exceeds server limit of 250")
+		require_Equal(t, err.Error(), "nats: consumer max request batch exceeds server limit of 250")
 
 		ci, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "dur2", AckPolicy: nats.AckExplicitPolicy, MaxAckPending: 500})
 		require_NoError(t, err)
@@ -16725,11 +16733,11 @@ func TestJetStreamLimits(t *testing.T) {
 
 		_, err = js.UpdateConsumer("foo", &nats.ConsumerConfig{Durable: "dur2", AckPolicy: nats.AckExplicitPolicy, MaxAckPending: 2000})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "consumer max ack pending exceeds system limit of 1000")
+		require_Equal(t, err.Error(), "nats: consumer max ack pending exceeds system limit of 1000")
 
 		_, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "dur3", AckPolicy: nats.AckExplicitPolicy, MaxAckPending: 2000})
 		require_Error(t, err)
-		require_Equal(t, err.Error(), "consumer max ack pending exceeds system limit of 1000")
+		require_Equal(t, err.Error(), "nats: consumer max ack pending exceeds system limit of 1000")
 	}
 
 	t.Run("clustered", func(t *testing.T) {
@@ -18757,10 +18765,8 @@ func TestJetStreamMirrorUpdatesNotSupported(t *testing.T) {
 	require_NoError(t, err)
 
 	cfg.Mirror = nil
-	if _, err := js.UpdateStream(cfg); err == nil ||
-		!strings.Contains(NewJSStreamMirrorNotUpdatableError().Error(), err.Error()) {
-		t.Fatalf("Expected error %q, got %q", NewJSStreamMirrorNotUpdatableError(), err)
-	}
+	_, err = js.UpdateStream(cfg)
+	require_Error(t, err, NewJSStreamMirrorNotUpdatableError())
 }
 
 func TestJetStreamDirectGetBySubject(t *testing.T) {
@@ -18866,64 +18872,6 @@ func TestJetStreamDirectGetBySubject(t *testing.T) {
 	}
 }
 
-// v2.9 will move to direct gets, this tests that we autoset them and auto-promote older streams.
-// This is keyed off a config setting for MaxMsgsPerSubject.
-func TestJetStreamDirectGetAutoSet(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	if config := s.JetStreamConfig(); config != nil {
-		defer removeDir(t, config.StoreDir)
-	}
-	defer s.Shutdown()
-
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:              "KV",
-		Subjects:          []string{"key.*"},
-		MaxMsgsPerSubject: 1,
-	})
-	require_NoError(t, err)
-
-	_, err = js.Publish("key.22", []byte("22"))
-	require_NoError(t, err)
-
-	// Make sure direct get was auto turned on.
-	subj := fmt.Sprintf(JSDirectMsgGetT, "KV")
-	req := []byte(`{"last_by_subj": "key.22"}`)
-	m, err := nc.Request(subj, req, time.Second)
-	require_NoError(t, err)
-	require_True(t, string(m.Data) == "22")
-
-	// New direct with key in subject.
-	subj = fmt.Sprintf(JSDirectGetLastBySubjectT, "KV", "key.22")
-	m, err = nc.Request(subj, nil, time.Second)
-	require_NoError(t, err)
-	require_True(t, string(m.Data) == "22")
-
-	// Make sure mirrors inherit.
-	_, err = js.AddStream(&nats.StreamConfig{
-		Name:   "M",
-		Mirror: &nats.StreamSource{Name: "KV"},
-	})
-	require_NoError(t, err)
-
-	// Now make sure mirrors are doing right thing with new way as well.
-	var sawMirror bool
-	subj = fmt.Sprintf(JSDirectGetLastBySubjectT, "KV", "key.22")
-	for i := 0; i < 100; i++ {
-		m, err := nc.Request(subj, nil, time.Second)
-		require_NoError(t, err)
-		if shdr := m.Header.Get(JSStream); shdr == "M" {
-			sawMirror = true
-			break
-		}
-	}
-	if !sawMirror {
-		t.Fatalf("Expected to see the mirror respond at least once")
-	}
-}
-
 func TestJetStreamProperErrorDueToOverlapSubjects(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	if config := s.JetStreamConfig(); config != nil {
@@ -19440,5 +19388,38 @@ func TestJetStreamStreamSubjectsOverlap(t *testing.T) {
 	})
 	require_Error(t, err)
 	require_True(t, strings.Contains(err.Error(), "overlaps"))
+}
+
+func TestJetStreamSuppressAllowDirect(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
 
+	si, err := js.AddStream(&nats.StreamConfig{
+		Name:              "TEST",
+		Subjects:          []string{"key.*"},
+		MaxMsgsPerSubject: 1,
+		AllowDirect:       true,
+	})
+	require_NoError(t, err)
+	require_True(t, si.Config.AllowDirect)
+
+	si, err = js.UpdateStream(&nats.StreamConfig{
+		Name:              "TEST",
+		Subjects:          []string{"key.*"},
+		MaxMsgsPerSubject: 1,
+		AllowDirect:       false,
+	})
+	require_NoError(t, err)
+	require_False(t, si.Config.AllowDirect)
+
+	sendStreamMsg(t, nc, "key.22", "msg")
+
+	_, err = js.GetLastMsg("TEST", "foo", nats.DirectGet(), nats.MaxWait(100*time.Millisecond))
+	require_Error(t, err)
 }
diff --git a/server/jwt_test.go b/server/jwt_test.go
index 251dfd6d..15fa4f71 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -5514,7 +5514,7 @@ func TestJWTJetStreamTiers(t *testing.T) {
 	// Test exceeding tiered stream limit
 	_, err = js.AddStream(&nats.StreamConfig{Name: "testR1-3", Replicas: 1, Subjects: []string{"testR1-3"}})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum number of streams reached")
+	require_Equal(t, err.Error(), "nats: maximum number of streams reached")
 
 	// Test tiers up to consumer limits
 	_, err = js.AddConsumer("testR1-1", &nats.ConsumerConfig{Durable: "dur1", AckPolicy: nats.AckExplicitPolicy})
@@ -5525,10 +5525,10 @@ func TestJWTJetStreamTiers(t *testing.T) {
 	// test exceeding tiered consumer limits
 	_, err = js.AddConsumer("testR1-1", &nats.ConsumerConfig{Durable: "dur4", AckPolicy: nats.AckExplicitPolicy})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum consumers limit reached")
+	require_Equal(t, err.Error(), "nats: maximum consumers limit reached")
 	_, err = js.AddConsumer("testR1-1", &nats.ConsumerConfig{Durable: "dur5", AckPolicy: nats.AckExplicitPolicy})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum consumers limit reached")
+	require_Equal(t, err.Error(), "nats: maximum consumers limit reached")
 
 	// test tiered storage limit
 	msg := [512]byte{}
@@ -5553,12 +5553,12 @@ func TestJWTJetStreamTiers(t *testing.T) {
 	require_NoError(t, err)
 	_, err = js.AddStream(&nats.StreamConfig{Name: "testR1-4", Replicas: 1, Subjects: []string{"testR1-4"}})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum number of streams reached")
+	require_Equal(t, err.Error(), "nats: maximum number of streams reached")
 	_, err = js.AddConsumer("testR1-3", &nats.ConsumerConfig{Durable: "dur6", AckPolicy: nats.AckExplicitPolicy})
 	require_NoError(t, err)
 	_, err = js.AddConsumer("testR1-3", &nats.ConsumerConfig{Durable: "dur7", AckPolicy: nats.AckExplicitPolicy})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "maximum consumers limit reached")
+	require_Equal(t, err.Error(), "nats: maximum consumers limit reached")
 	_, err = js.Publish("testR1-3", msg[:])
 	require_NoError(t, err)
 	_, err = js.Publish("testR1-3", []byte("1"))
@@ -5624,7 +5624,7 @@ func TestJWTJetStreamMaxAckPending(t *testing.T) {
 	_, err = js.AddConsumer("foo", &nats.ConsumerConfig{
 		Durable: "dur1", AckPolicy: nats.AckAllPolicy, MaxAckPending: 2000})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "consumer max ack pending exceeds system limit of 1000")
+	require_Equal(t, err.Error(), "nats: consumer max ack pending exceeds system limit of 1000")
 
 	ci, err := js.AddConsumer("foo", &nats.ConsumerConfig{
 		Durable: "dur2", AckPolicy: nats.AckAllPolicy, MaxAckPending: 500})
@@ -5634,7 +5634,7 @@ func TestJWTJetStreamMaxAckPending(t *testing.T) {
 	_, err = js.UpdateConsumer("foo", &nats.ConsumerConfig{
 		Durable: "dur2", AckPolicy: nats.AckAllPolicy, MaxAckPending: 2000})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "consumer max ack pending exceeds system limit of 1000")
+	require_Equal(t, err.Error(), "nats: consumer max ack pending exceeds system limit of 1000")
 
 	time.Sleep(time.Second - time.Since(start)) // make sure the time stamp changes
 	accClaim.Limits.JetStreamTieredLimits["R1"] = jwt.JetStreamLimits{
@@ -5703,7 +5703,7 @@ func TestJWTJetStreamMaxStreamBytes(t *testing.T) {
 
 	_, err = js.AddStream(&nats.StreamConfig{Name: "foo", Replicas: 1, MaxBytes: 2048})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "stream max bytes exceeds account limit max stream bytes")
+	require_Equal(t, err.Error(), "nats: stream max bytes exceeds account limit max stream bytes")
 	_, err = js.AddStream(&nats.StreamConfig{Name: "foo", Replicas: 1, MaxBytes: 1024})
 	require_NoError(t, err)
 
@@ -5725,7 +5725,7 @@ func TestJWTJetStreamMaxStreamBytes(t *testing.T) {
 
 	_, err = js.AddStream(&nats.StreamConfig{Name: "bar", Replicas: 1, MaxBytes: 3000})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "stream max bytes exceeds account limit max stream bytes")
+	require_Equal(t, err.Error(), "nats: stream max bytes exceeds account limit max stream bytes")
 	_, err = js.AddStream(&nats.StreamConfig{Name: "bar", Replicas: 1, MaxBytes: 2048})
 	require_NoError(t, err)
 
@@ -5739,7 +5739,7 @@ func TestJWTJetStreamMaxStreamBytes(t *testing.T) {
 	// test disabling max bytes required
 	_, err = js.UpdateStream(&nats.StreamConfig{Name: "bar", Replicas: 1})
 	require_Error(t, err)
-	require_Equal(t, err.Error(), "account requires a stream config to have max bytes set")
+	require_Equal(t, err.Error(), "nats: account requires a stream config to have max bytes set")
 }
 
 func TestJWTQueuePermissions(t *testing.T) {
diff --git a/server/norace_test.go b/server/norace_test.go
index 1b3eee12..ecf36fd4 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -3344,7 +3344,11 @@ func TestNoRaceJetStreamClusterInterestPolicyAckNone(t *testing.T) {
 				atomic.AddUint32(&received, 1)
 			}
 
-			_, err = js.Subscribe("cluster.created", mh, nats.Durable(test.durable), nats.DeliverNew(), nats.AckNone())
+			opts := []nats.SubOpt{nats.DeliverNew(), nats.AckNone()}
+			if test.durable != _EMPTY_ {
+				opts = append(opts, nats.Durable(test.durable))
+			}
+			_, err = js.Subscribe("cluster.created", mh, opts...)
 			if err != nil {
 				t.Fatalf("Unexepected error: %v", err)
 			}
diff --git a/server/sublist_test.go b/server/sublist_test.go
index ed954a78..668fcaaa 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -1125,7 +1125,7 @@ func TestSublistRegisterInterestNotification(t *testing.T) {
 	}
 
 	tt := time.NewTimer(time.Second)
-	expectBoolWithCh := func(ch chan bool, b bool) {
+	expectBool := func(b bool) {
 		t.Helper()
 		tt.Reset(time.Second)
 		defer tt.Stop()
@@ -1138,10 +1138,6 @@ func TestSublistRegisterInterestNotification(t *testing.T) {
 			t.Fatalf("Timeout waiting for expected value")
 		}
 	}
-	expectBool := func(b bool) {
-		t.Helper()
-		expectBoolWithCh(ch, b)
-	}
 	expectFalse := func() {
 		t.Helper()
 		expectBool(false)
@@ -1156,16 +1152,12 @@ func TestSublistRegisterInterestNotification(t *testing.T) {
 			t.Fatalf("Expected no notifications, had %d and first was %v", lch, <-ch)
 		}
 	}
-	expectOneWithCh := func(ch chan bool) {
+	expectOne := func() {
 		t.Helper()
 		if len(ch) != 1 {
 			t.Fatalf("Expected 1 notification")
 		}
 	}
-	expectOne := func() {
-		t.Helper()
-		expectOneWithCh(ch)
-	}
 
 	expectOne()
 	expectFalse()
@@ -1356,43 +1348,6 @@ func TestSublistRegisterInterestNotification(t *testing.T) {
 		t.Fatalf("Expected to return true")
 	}
 
-	if err := s.RegisterQueueNotification("some.subject", "queue1", ch); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	expectOne()
-	expectFalse()
-
-	qsub1 = newQSub("some.subject", "queue1")
-	s.Insert(qsub1)
-	expectTrue()
-
-	// Create a second channel for this other queue
-	ch2 := make(chan bool, 1)
-	if err := s.RegisterQueueNotification("some.subject", "queue2", ch2); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	expectOneWithCh(ch2)
-	expectBoolWithCh(ch2, false)
-
-	qsub2 = newQSub("some.subject", "queue2")
-	s.Insert(qsub2)
-	expectBoolWithCh(ch2, true)
-
-	// But we should not get notification on queue1
-	expectNone()
-
-	s.Remove(qsub1)
-	expectFalse()
-	s.Remove(qsub2)
-	expectBoolWithCh(ch2, false)
-
-	if !s.ClearQueueNotification("some.subject", "queue1", ch) {
-		t.Fatalf("Expected to return true")
-	}
-	if !s.ClearQueueNotification("some.subject", "queue2", ch2) {
-		t.Fatalf("Expected to return true")
-	}
-
 	// Test non-blocking notifications.
 	if err := s.RegisterNotification("bar", ch); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
diff --git a/server/test_test.go b/server/test_test.go
index 9cb4c0c2..8b757193 100644
--- a/server/test_test.go
+++ b/server/test_test.go
@@ -98,8 +98,15 @@ func require_Error(t *testing.T, err error, expected ...error) {
 	if len(expected) == 0 {
 		return
 	}
+	// Try to strip nats prefix from Go library if present.
+	const natsErrPre = "nats: "
+	eStr := err.Error()
+	if strings.HasPrefix(eStr, natsErrPre) {
+		eStr = strings.Replace(eStr, natsErrPre, _EMPTY_, 1)
+	}
+
 	for _, e := range expected {
-		if err == e || strings.Contains(e.Error(), err.Error()) {
+		if err == e || strings.Contains(e.Error(), eStr) {
 			return
 		}
 	}