Repository: nats-io/nats.go
Pull request #1885 (2025-06-20 15:05:10)

[FIXED] Ordered consumer not closing on connection close
Ordered consumers were not properly closing when the underlying connection was closed.

Signed-off-by: Piotr Piotrowski [piotr@synadia.com](mailto:piotr@synadia.com)

coveralls: 
[![Coverage Status](https://coveralls.io/builds/74265811/badge)](https://coveralls.io/builds/74265811)

coverage: 84.741% (-0.02%) from 84.76%
when pulling **e091e95e1dc990e0ca3719930781c0be4cebde21 on fix-ordered-consumer-closed**
into **08798499517c45ece8bfb3440b939d079ef1b0d4 on main**.


Patch:
diff --git a/jetstream/api.go b/jetstream/api.go
index 0fc5d91..0e010f9 100644
--- a/jetstream/api.go
+++ b/jetstream/api.go
@@ -127,8 +127,8 @@ func (js *jetStream) apiRequest(ctx context.Context, subj string, data ...[]byte
 	if len(data) > 0 {
 		req = data[0]
 	}
-	if js.opts.clientTrace != nil {
-		ctrace := js.opts.clientTrace
+	if js.opts.ClientTrace != nil {
+		ctrace := js.opts.ClientTrace
 		if ctrace.RequestSent != nil {
 			ctrace.RequestSent(subj, req)
 		}
@@ -137,8 +137,8 @@ func (js *jetStream) apiRequest(ctx context.Context, subj string, data ...[]byte
 	if err != nil {
 		return nil, err
 	}
-	if js.opts.clientTrace != nil {
-		ctrace := js.opts.clientTrace
+	if js.opts.ClientTrace != nil {
+		ctrace := js.opts.ClientTrace
 		if ctrace.ResponseReceived != nil {
 			ctrace.ResponseReceived(subj, resp.Data, resp.Header)
 		}
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index 960febe..eabfec0 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -315,6 +315,9 @@ type (
 		// a deadline set.
 		DefaultTimeout time.Duration
 
+		// ClientTrace enables request/response API calls tracing.
+		ClientTrace *ClientTrace
+
 		publisherOpts asyncPublisherOpts
 
 		// this is the actual prefix used in the API requests
@@ -322,7 +325,6 @@ type (
 		apiPrefix      string
 		replyPrefix    string
 		replyPrefixLen int
-		clientTrace    *ClientTrace
 	}
 
 	// ClientTrace can be used to trace API interactions for [JetStream].
@@ -528,7 +530,13 @@ func (js *jetStream) Conn() *nats.Conn {
 }
 
 func (js *jetStream) Options() JetStreamOptions {
-	return js.opts
+	opts := js.opts
+	// Return a copy of ClientTrace to prevent modification
+	if opts.ClientTrace != nil {
+		clientTraceCopy := *opts.ClientTrace
+		opts.ClientTrace = &clientTraceCopy
+	}
+	return opts
 }
 
 // CreateStream creates a new stream with given config and returns an
diff --git a/jetstream/jetstream_options.go b/jetstream/jetstream_options.go
index 1baa0e4..fbd65fb 100644
--- a/jetstream/jetstream_options.go
+++ b/jetstream/jetstream_options.go
@@ -31,7 +31,7 @@ func (fn pullOptFunc) configureMessages(opts *consumeOpts) error {
 // WithClientTrace enables request/response API calls tracing.
 func WithClientTrace(ct *ClientTrace) JetStreamOpt {
 	return func(opts *JetStreamOptions) error {
-		opts.clientTrace = ct
+		opts.ClientTrace = ct
 		return nil
 	}
 }
diff --git a/jetstream/kv.go b/jetstream/kv.go
index f2c1405..8cb7f2d 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -834,10 +834,10 @@ func (js *jetStream) legacyJetStream() (nats.JetStreamContext, error) {
 	if js.opts.apiPrefix != "" {
 		opts = append(opts, nats.APIPrefix(js.opts.apiPrefix))
 	}
-	if js.opts.clientTrace != nil {
+	if js.opts.ClientTrace != nil {
 		opts = append(opts, nats.ClientTrace{
-			RequestSent:      js.opts.clientTrace.RequestSent,
-			ResponseReceived: js.opts.clientTrace.ResponseReceived,
+			RequestSent:      js.opts.ClientTrace.RequestSent,
+			ResponseReceived: js.opts.ClientTrace.ResponseReceived,
 		})
 	}
 	return js.conn.JetStream(opts...)
diff --git a/jetstream/ordered.go b/jetstream/ordered.go
index 8d79bc4..138145f 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -80,6 +80,8 @@ var (
 //
 // See [Consumer.Consume] for more details.
 func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (ConsumeContext, error) {
+	c.Lock()
+	defer c.Unlock()
 	if (c.consumerType == consumerTypeNotSet || c.consumerType == consumerTypeConsume) && c.currentConsumer == nil {
 		err := c.reset()
 		if err != nil {
@@ -201,10 +203,20 @@ func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt
 func (c *orderedConsumer) errHandler(serial int) func(cc ConsumeContext, err error) {
 	return func(cc ConsumeContext, err error) {
 		c.Lock()
-		defer c.Unlock()
+
 		if c.userErrHandler != nil && !errors.Is(err, errOrderedSequenceMismatch) && !errors.Is(err, errConnected) {
 			c.userErrHandler(cc, err)
 		}
+		if errors.Is(err, ErrConnectionClosed) {
+			if c.subscription != nil {
+				c.Unlock()
+				c.subscription.Stop()
+				return
+			}
+			c.Unlock()
+			return
+		}
+
 		if errors.Is(err, ErrNoHeartbeat) ||
 			errors.Is(err, errOrderedSequenceMismatch) ||
 			errors.Is(err, ErrConsumerDeleted) ||
@@ -216,6 +228,7 @@ func (c *orderedConsumer) errHandler(serial int) func(cc ConsumeContext, err err
 				c.doReset <- struct{}{}
 			}
 		}
+		c.Unlock()
 	}
 }
 
@@ -359,29 +372,23 @@ func (s *orderedSubscription) Drain() {
 // fully stopped/drained. When the channel is closed, no more messages
 // will be received and processing is complete.
 func (s *orderedSubscription) Closed() <-chan struct{} {
-	s.consumer.Lock()
-	defer s.consumer.Unlock()
 	closedCh := make(chan struct{})
 
 	go func() {
-		for {
-			s.consumer.Lock()
-			if s.consumer.currentSub == nil {
-				return
-			}
+		// First wait for s.done to be closed
+		<-s.done
 
+		// Then ensure underlying consumer is also closed (it may still be draining)
+		s.consumer.Lock()
+		if s.consumer.currentSub != nil {
 			closed := s.consumer.currentSub.Closed()
 			s.consumer.Unlock()
-
-			// wait until the underlying pull consumer is closed
 			<-closed
-			// if the subscription is closed and ordered consumer is closed as well,
-			// send a signal that the Consume() is fully stopped
-			if s.closed.Load() == 1 {
-				close(closedCh)
-				return
-			}
+		} else {
+			s.consumer.Unlock()
 		}
+
+		close(closedCh)
 	}()
 	return closedCh
 }
diff --git a/jetstream/stream.go b/jetstream/stream.go
index 42a5f11..c1db7ce 100644
--- a/jetstream/stream.go
+++ b/jetstream/stream.go
@@ -138,7 +138,7 @@ type (
 	StreamInfoOpt func(*streamInfoRequest) error
 
 	streamInfoRequest struct {
-		apiPaged
+		apiPagedRequest
 		DeletedDetails bool   `json:"deleted_details,omitempty"`
 		SubjectFilter  string `json:"subjects_filter,omitempty"`
 	}
diff --git a/kv.go b/kv.go
index e1ab3c0..332a5cc 100644
--- a/kv.go
+++ b/kv.go
@@ -115,12 +115,22 @@ type KeyWatcher interface {
 	Updates() <-chan KeyValueEntry
 	// Stop will stop this watcher.
 	Stop() error
+	// Error returns a channel that will receive any error that occurs during
+	// watching. In particular, this will receive an error if the watcher times
+	// out while expecting more initial keys. The channel is closed when the
+	// watch operation completes or when Stop() is called.
+	Error() <-chan error
 }
 
 // KeyLister is used to retrieve a list of key value store keys
 type KeyLister interface {
 	Keys() <-chan string
 	Stop() error
+	// Error returns a channel that will receive any error that occurs during
+	// key listing. In particular, this will receive an error if the underlying
+	// watcher times out while expecting more keys. The channel is closed when
+	// the listing operation completes or when Stop() is called.
+	Error() <-chan error
 }
 
 type WatchOpt interface {
@@ -331,6 +341,7 @@ var (
 	ErrKeyDeleted             = errors.New("nats: key was deleted")
 	ErrHistoryToLarge         = errors.New("nats: history limited to a max of 64")
 	ErrNoKeysFound            = errors.New("nats: no keys found")
+	ErrKeyWatcherTimeout      = errors.New("nats: key watcher timed out waiting for initial keys")
 )
 
 var (
@@ -911,6 +922,10 @@ func (kl *keyLister) Stop() error {
 	return kl.watcher.Stop()
 }
 
+func (kl *keyLister) Error() <-chan error {
+	return kl.watcher.Error()
+}
+
 // History will return all values for the key.
 func (kv *kvs) History(key string, opts ...WatchOpt) ([]KeyValueEntry, error) {
 	opts = append(opts, IncludeHistory())
@@ -943,6 +958,7 @@ type watcher struct {
 	received      uint64
 	ctx           context.Context
 	initDoneTimer *time.Timer
+	errCh         chan error
 }
 
 // Context returns the context for the watcher if set.
@@ -969,6 +985,16 @@ func (w *watcher) Stop() error {
 	return w.sub.Unsubscribe()
 }
 
+// Error returns a channel that will receive any error that occurs during watching.
+func (w *watcher) Error() <-chan error {
+	if w == nil {
+		closedCh := make(chan error)
+		close(closedCh)
+		return closedCh
+	}
+	return w.errCh
+}
+
 // WatchAll watches all keys.
 func (kv *kvs) WatchAll(opts ...WatchOpt) (KeyWatcher, error) {
 	return kv.Watch(AllKeys, opts...)
@@ -1006,7 +1032,11 @@ func (kv *kvs) WatchFiltered(keys []string, opts ...WatchOpt) (KeyWatcher, error
 	}
 
 	// We will block below on placing items on the chan. That is by design.
-	w := &watcher{updates: make(chan KeyValueEntry, 256), ctx: o.ctx}
+	w := &watcher{
+		updates: make(chan KeyValueEntry, 256),
+		ctx:     o.ctx,
+		errCh:   make(chan error, 1),
+	}
 
 	update := func(m *Msg) {
 		tokens, err := parser.GetMetadataFields(m.Reply)
@@ -1108,6 +1138,10 @@ func (kv *kvs) WatchFiltered(keys []string, opts ...WatchOpt) (KeyWatcher, error
 				defer w.mu.Unlock()
 				if !w.initDone {
 					w.initDone = true
+					select {
+					case w.errCh <- ErrKeyWatcherTimeout:
+					default:
+					}
 					w.updates <- nil
 				}
 			})
@@ -1124,6 +1158,7 @@ func (kv *kvs) WatchFiltered(keys []string, opts ...WatchOpt) (KeyWatcher, error
 			w.initDoneTimer.Stop()
 		}
 		close(w.updates)
+		close(w.errCh)
 	}
 
 	sub.mu.Unlock()
diff --git a/micro/service.go b/micro/service.go
index 54f6f1d..1d4ad46 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -74,6 +74,8 @@ type (
 		metadata   map[string]string
 		queueGroup string
 		qgDisabled bool
+		msgLimit   int
+		bytesLimit int
 	}
 
 	groupOpts struct {
@@ -222,6 +224,7 @@ type (
 	NATSError struct {
 		Subject     string
 		Description string
+		err         error
 	}
 
 	// service represents a configured NATS service.
@@ -373,7 +376,9 @@ func AddService(nc *nats.Conn, config Config) (Service, error) {
 			response, _ := json.Marshal(valuef())
 			if err := req.Respond(response); err != nil {
 				if err := req.Error("500", fmt.Sprintf("Error handling %s request: %s", verb, err), nil); err != nil && config.ErrorHandler != nil {
-					svc.asyncDispatcher.push(func() { config.ErrorHandler(svc, &NATSError{req.Subject(), err.Error()}) })
+					svc.asyncDispatcher.push(func() {
+						config.ErrorHandler(svc, &NATSError{Subject: req.Subject(), Description: err.Error(), err: err})
+					})
 				}
 			}
 		}
@@ -407,10 +412,10 @@ func (s *service) AddEndpoint(name string, handler Handler, opts ...EndpointOpt)
 		subject = options.subject
 	}
 	queueGroup, noQueue := resolveQueueGroup(options.queueGroup, s.Config.QueueGroup, options.qgDisabled, s.Config.QueueGroupDisabled)
-	return addEndpoint(s, name, subject, handler, options.metadata, queueGroup, noQueue)
+	return addEndpoint(s, name, subject, handler, options.metadata, queueGroup, noQueue, options.msgLimit, options.bytesLimit)
 }
 
-func addEndpoint(s *service, name, subject string, handler Handler, metadata map[string]string, queueGroup string, noQueue bool) error {
+func addEndpoint(s *service, name, subject string, handler Handler, metadata map[string]string, queueGroup string, noQueue bool, msgLimit, bytesLimit int) error {
 	if !nameRegexp.MatchString(name) {
 		return fmt.Errorf("%w: invalid endpoint name", ErrConfigValidation)
 	}
@@ -434,6 +439,10 @@ func addEndpoint(s *service, name, subject string, handler Handler, metadata map
 
 	var sub *nats.Subscription
 	var err error
+	var options = endpointOpts{
+		msgLimit:   msgLimit,
+		bytesLimit: bytesLimit,
+	}
 
 	if !noQueue {
 		sub, err = s.nc.QueueSubscribe(
@@ -454,6 +463,14 @@ func addEndpoint(s *service, name, subject string, handler Handler, metadata map
 	if err != nil {
 		return err
 	}
+
+	// Apply pending limits if configured
+	if options.msgLimit != 0 || options.bytesLimit != 0 {
+		if err := sub.SetPendingLimits(options.msgLimit, options.bytesLimit); err != nil {
+			return err
+		}
+	}
+
 	s.m.Lock()
 	endpoint.subscription = sub
 	s.endpoints = append(s.endpoints, endpoint)
@@ -549,6 +566,7 @@ func (s *service) wrapConnectionEventCallbacks() {
 				s.Config.ErrorHandler(s, &NATSError{
 					Subject:     sub.Subject,
 					Description: err.Error(),
+					err:         err,
 				})
 			}
 			s.m.Lock()
@@ -576,6 +594,7 @@ func (s *service) wrapConnectionEventCallbacks() {
 				s.Config.ErrorHandler(s, &NATSError{
 					Subject:     sub.Subject,
 					Description: err.Error(),
+					err:         err,
 				})
 			}
 			s.m.Lock()
@@ -807,6 +826,22 @@ func (e *NATSError) Error() string {
 	return fmt.Sprintf("%q: %s", e.Subject, e.Description)
 }
 
+// Unwrap returns the underlying error if any.
+func (e *NATSError) Unwrap() error {
+	return e.err
+}
+
+// Is reports whether the target error is equal to this error.
+func (e *NATSError) Is(target error) bool {
+	if e == nil {
+		return false
+	}
+	if t, ok := target.(*NATSError); ok {
+		return e.Subject == t.Subject && e.Description == t.Description
+	}
+	return e.err != nil && errors.Is(e.err, target)
+}
+
 func (g *group) AddEndpoint(name string, handler Handler, opts ...EndpointOpt) error {
 	var options endpointOpts
 	for _, opt := range opts {
@@ -824,7 +859,7 @@ func (g *group) AddEndpoint(name string, handler Handler, opts ...EndpointOpt) e
 	}
 	queueGroup, noQueue := resolveQueueGroup(options.queueGroup, g.queueGroup, options.qgDisabled, g.queueGroupDisabled)
 
-	return addEndpoint(g.service, name, endpointSubject, handler, options.metadata, queueGroup, noQueue)
+	return addEndpoint(g.service, name, endpointSubject, handler, options.metadata, queueGroup, noQueue, options.msgLimit, options.bytesLimit)
 }
 
 func resolveQueueGroup(customQG, parentQG string, disabled, parentDisabled bool) (string, bool) {
@@ -944,6 +979,21 @@ func WithEndpointQueueGroupDisabled() EndpointOpt {
 	}
 }
 
+// WithEndpointPendingLimits sets the pending limits for the endpoint's
+// subscription. These limits how many messages and/or bytes can be buffered in
+// memory before the subscription is terminated with nats.ErrSlowConsumer.
+// Either limit can be set to -1 to indicate no limit.
+func WithEndpointPendingLimits(msgLimit, bytesLimit int) EndpointOpt {
+	return func(e *endpointOpts) error {
+		if msgLimit == 0 && bytesLimit == 0 {
+			return fmt.Errorf("%w: at least one pending limit must be non-zero", ErrConfigValidation)
+		}
+		e.msgLimit = msgLimit
+		e.bytesLimit = bytesLimit
+		return nil
+	}
+}
+
 func WithGroupQueueGroup(queueGroup string) GroupOpt {
 	return func(g *groupOpts) {
 		g.queueGroup = queueGroup
diff --git a/nats.go b/nats.go
index 20833b5..e48c93b 100644
--- a/nats.go
+++ b/nats.go
@@ -1247,6 +1247,35 @@ func UserCredentials(userOrChainedFile string, seedFiles ...string) Option {
 	return UserJWT(userCB, sigCB)
 }
 
+// UserCredentialBytes is a convenience function that takes the JWT and seed
+// values as byte slices. This allows passing credentials directly from memory
+// or environment variables without needing to write them to disk.
+func UserCredentialBytes(userOrChainedFileBytes []byte, seedFiles ...[]byte) Option {
+	userCB := func() (string, error) {
+		return nkeys.ParseDecoratedJWT(userOrChainedFileBytes)
+	}
+
+	var seedBytes []byte
+	if len(seedFiles) > 0 {
+		seedBytes = seedFiles[0]
+	} else {
+		seedBytes = userOrChainedFileBytes
+	}
+
+	sigCB := func(nonce []byte) ([]byte, error) {
+		kp, err := nkeys.ParseDecoratedNKey(seedBytes)
+		if err != nil {
+			return nil, fmt.Errorf("unable to extract key pair from bytes: %w", err)
+		}
+		defer kp.Wipe()
+
+		sig, _ := kp.Sign(nonce)
+		return sig, nil
+	}
+
+	return UserJWT(userCB, sigCB)
+}
+
 // UserJWTAndSeed is a convenience function that takes the JWT and seed
 // values as strings.
 func UserJWTAndSeed(jwt string, seed string) Option {
@@ -6120,18 +6149,12 @@ func sigHandler(nonce []byte, seedFile string) ([]byte, error) {
 type timeoutWriter struct {
 	timeout time.Duration
 	conn    net.Conn
-	err     error
 }
 
 // Write implements the io.Writer interface.
 func (tw *timeoutWriter) Write(p []byte) (int, error) {
-	if tw.err != nil {
-		return 0, tw.err
-	}
-
-	var n int
 	tw.conn.SetWriteDeadline(time.Now().Add(tw.timeout))
-	n, tw.err = tw.conn.Write(p)
+	n, err := tw.conn.Write(p)
 	tw.conn.SetWriteDeadline(time.Time{})
-	return n, tw.err
+	return n, err
 }


Test patch:
diff --git a/jetstream/test/jetstream_test.go b/jetstream/test/jetstream_test.go
index 86addad..17d86c1 100644
--- a/jetstream/test/jetstream_test.go
+++ b/jetstream/test/jetstream_test.go
@@ -188,10 +188,20 @@ func TestJetStreamOptionsReadOnly(t *testing.T) {
 	}
 	defer nc.Close()
 
-	js, err := jetstream.New(nc)
+	// Test with ClientTrace
+	var originalCalled bool
+	clientTrace := &jetstream.ClientTrace{
+		RequestSent: func(subj string, payload []byte) {
+			originalCalled = true
+		},
+	}
+
+	js, err := jetstream.New(nc, jetstream.WithClientTrace(clientTrace))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
+
+	// Test that modifying basic options doesn't affect the original
 	opts := js.Options()
 	opts.APIPrefix = "foo"
 	opts.Domain = "bar"
@@ -203,6 +213,32 @@ func TestJetStreamOptionsReadOnly(t *testing.T) {
 	if opts.Domain != "" {
 		t.Fatalf("Invalid domain; want: %v, got: %v", "", opts.Domain)
 	}
+
+	if opts.ClientTrace == nil {
+		t.Fatal("Expected ClientTrace to be set in Options()")
+	}
+	if opts.ClientTrace == clientTrace {
+		t.Fatal("ClientTrace should be a copy, not the same instance")
+	}
+
+	var modifiedCalled bool
+	opts.ClientTrace.RequestSent = func(subj string, payload []byte) {
+		modifiedCalled = true
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	_, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: "test", Subjects: []string{"test.*"}})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	if !originalCalled {
+		t.Fatal("Original callback should have been called")
+	}
+	if modifiedCalled {
+		t.Fatal("Modified callback should not have been called")
+	}
 }
 
 func TestWithClientTrace(t *testing.T) {
@@ -216,14 +252,15 @@ func TestWithClientTrace(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
 	defer cancel()
 	var sent, received string
-	js, err := jetstream.New(nc, jetstream.WithClientTrace(&jetstream.ClientTrace{
+	clientTrace := &jetstream.ClientTrace{
 		RequestSent: func(subj string, _ []byte) {
 			sent = fmt.Sprintf("Request sent: %s", subj)
 		},
 		ResponseReceived: func(subj string, _ []byte, _ nats.Header) {
 			received = fmt.Sprintf("Response received: %s", subj)
 		},
-	}))
+	}
+	js, err := jetstream.New(nc, jetstream.WithClientTrace(clientTrace))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
diff --git a/jetstream/test/ordered_test.go b/jetstream/test/ordered_test.go
index f408162..2837702 100644
--- a/jetstream/test/ordered_test.go
+++ b/jetstream/test/ordered_test.go
@@ -2121,3 +2121,57 @@ func TestOrderedConsumerConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestOrderedConsumerCloseConn(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer nc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	consumer, err := s.OrderedConsumer(context.Background(), jetstream.OrderedConsumerConfig{})
+	if err != nil {
+		t.Fatalf("Failed to create ordered consumer: %v", err)
+	}
+
+	gotConnClosedErr := make(chan struct{})
+
+	time.AfterFunc(500*time.Millisecond, func() {
+		nc.Close()
+	})
+
+	oc, err := consumer.Consume(func(msg jetstream.Msg) {
+	}, jetstream.ConsumeErrHandler(func(consumeCtx jetstream.ConsumeContext, err error) {
+		if errors.Is(err, jetstream.ErrConnectionClosed) {
+			close(gotConnClosedErr)
+		}
+	}))
+	if err != nil {
+		t.Fatalf("Failed to consume: %v", err)
+	}
+
+	select {
+	case <-gotConnClosedErr:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timeout waiting for connection closed error")
+	}
+
+	select {
+	case <-oc.Closed():
+	case <-time.After(2 * time.Second):
+		t.Fatalf("Timeout waiting for close")
+	}
+}
diff --git a/micro/error_test.go b/micro/error_test.go
new file mode 100644
index 0000000..88adc74
--- /dev/null
+++ b/micro/error_test.go
@@ -0,0 +1,106 @@
+// Copyright 2025 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 micro
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestNATSErrorUnwrap(t *testing.T) {
+	baseErr := errors.New("base error")
+	natsErr := &NATSError{
+		Subject:     "test.subject",
+		Description: "test description",
+		err:         baseErr,
+	}
+
+	unwrapped := errors.Unwrap(natsErr)
+	if unwrapped != baseErr {
+		t.Errorf("Unwrap() = %v, want %v", unwrapped, baseErr)
+	}
+
+	// Test with nil underlying error
+	natsErr2 := &NATSError{
+		Subject:     "test.subject",
+		Description: "test description",
+	}
+	if errors.Unwrap(natsErr2) != nil {
+		t.Errorf("Unwrap() with nil err should return nil")
+	}
+}
+
+func TestNATSErrorIs(t *testing.T) {
+	natsErr1 := &NATSError{
+		Subject:     "test.subject",
+		Description: "test description",
+	}
+
+	natsErr2 := &NATSError{
+		Subject:     "test.subject",
+		Description: "test description",
+	}
+
+	natsErr3 := &NATSError{
+		Subject:     "different.subject",
+		Description: "test description",
+	}
+
+	natsErr4 := &NATSError{
+		Subject:     "test.subject",
+		Description: "different description",
+	}
+
+	// Test equality
+	if !errors.Is(natsErr1, natsErr2) {
+		t.Errorf("Is() = false, want true for equal errors")
+	}
+
+	// different subject
+	if errors.Is(natsErr1, natsErr3) {
+		t.Errorf("Is() = true, want false for different subjects")
+	}
+
+	// different description
+	if errors.Is(natsErr1, natsErr4) {
+		t.Errorf("Is() = true, want false for different descriptions")
+	}
+
+	// non NATSError
+	otherErr := errors.New("other error")
+	if errors.Is(natsErr1, otherErr) {
+		t.Errorf("Is() = true, want false for non-NATSError")
+	}
+}
+
+func TestNATSErrorWrapping(t *testing.T) {
+	// Test error chain
+	baseErr := errors.New("connection failed")
+	natsErr := &NATSError{
+		Subject:     "test.subject",
+		Description: "service error: connection failed",
+		err:         baseErr,
+	}
+
+	// Check that we can find the base error in the chain
+	if !errors.Is(natsErr, baseErr) {
+		t.Errorf("errors.Is() should find base error in chain")
+	}
+
+	// Test error message
+	expectedMsg := `"test.subject": service error: connection failed`
+	if natsErr.Error() != expectedMsg {
+		t.Errorf("Error() = %q, want %q", natsErr.Error(), expectedMsg)
+	}
+}
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index ac3c638..a43b8ce 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -1910,3 +1910,183 @@ func TestDisableQueueGroup(t *testing.T) {
 	wg.Wait()
 
 }
+
+func TestEndpointPendingLimits(t *testing.T) {
+	s := RunServerOnPort(-1)
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL())
+	if err != nil {
+		t.Fatalf("Error connecting to server: %v", err)
+	}
+	defer nc.Close()
+
+	handler := micro.HandlerFunc(func(req micro.Request) {
+		time.Sleep(10 * time.Millisecond)
+		req.Respond([]byte("OK"))
+	})
+
+	t.Run("with pending limits", func(t *testing.T) {
+		srv, err := micro.AddService(nc, micro.Config{
+			Name:    "test-service",
+			Version: "1.0.0",
+		})
+		if err != nil {
+			t.Fatalf("Error adding service: %v", err)
+		}
+		defer srv.Stop()
+
+		msgLimit := 10
+		bytesLimit := 1024
+
+		err = srv.AddEndpoint("test", handler,
+			micro.WithEndpointSubject("test.pending"),
+			micro.WithEndpointPendingLimits(msgLimit, bytesLimit))
+		if err != nil {
+			t.Fatalf("Error adding endpoint: %v", err)
+		}
+
+		// Verify endpoint was created
+		info := srv.Info()
+		if len(info.Endpoints) != 1 {
+			t.Fatalf("Expected 1 endpoint, got %d", len(info.Endpoints))
+		}
+
+		// Send messages and verify they're handled
+		for i := range 5 {
+			resp, err := nc.Request("test.pending", []byte("test"), 100*time.Millisecond)
+			if err != nil {
+				t.Fatalf("Error on request %d: %v", i, err)
+			}
+			if string(resp.Data) != "OK" {
+				t.Fatalf("Expected OK response, got %s", string(resp.Data))
+			}
+		}
+	})
+
+	t.Run("with only msg limit", func(t *testing.T) {
+		srv, err := micro.AddService(nc, micro.Config{
+			Name:    "test-service-msg-only",
+			Version: "1.0.0",
+		})
+		if err != nil {
+			t.Fatalf("Error adding service: %v", err)
+		}
+		defer srv.Stop()
+
+		msgLimit := 50
+		bytesLimit := -1
+
+		err = srv.AddEndpoint("test", handler,
+			micro.WithEndpointSubject("test.msg.only"),
+			micro.WithEndpointPendingLimits(msgLimit, bytesLimit))
+		if err != nil {
+			t.Fatalf("Error adding endpoint: %v", err)
+		}
+
+		resp, err := nc.Request("test.msg.only", []byte("test"), 100*time.Millisecond)
+		if err != nil {
+			t.Fatalf("Error on request: %v", err)
+		}
+		if string(resp.Data) != "OK" {
+			t.Fatalf("Expected OK response, got %s", string(resp.Data))
+		}
+	})
+
+	t.Run("with only bytes limit", func(t *testing.T) {
+		srv, err := micro.AddService(nc, micro.Config{
+			Name:    "test-service-bytes-only",
+			Version: "1.0.0",
+		})
+		if err != nil {
+			t.Fatalf("Error adding service: %v", err)
+		}
+		defer srv.Stop()
+
+		msgLimit := -1
+		bytesLimit := 4096
+
+		err = srv.AddEndpoint("test", handler,
+			micro.WithEndpointSubject("test.bytes.only"),
+			micro.WithEndpointPendingLimits(msgLimit, bytesLimit))
+		if err != nil {
+			t.Fatalf("Error adding endpoint: %v", err)
+		}
+
+		resp, err := nc.Request("test.bytes.only", []byte("test"), 100*time.Millisecond)
+		if err != nil {
+			t.Fatalf("Error on request: %v", err)
+		}
+		if string(resp.Data) != "OK" {
+			t.Fatalf("Expected OK response, got %s", string(resp.Data))
+		}
+	})
+
+	t.Run("invalid limits", func(t *testing.T) {
+		srv, err := micro.AddService(nc, micro.Config{
+			Name:    "test-service-invalid",
+			Version: "1.0.0",
+		})
+		if err != nil {
+			t.Fatalf("Error adding service: %v", err)
+		}
+		defer srv.Stop()
+
+		err = srv.AddEndpoint("test", handler,
+			micro.WithEndpointPendingLimits(0, 0))
+		if !errors.Is(err, micro.ErrConfigValidation) {
+			t.Fatal("Expected error when both limits are zero")
+		}
+	})
+
+	t.Run("pending limits trigger slow consumer error", func(t *testing.T) {
+		errCh := make(chan error, 1)
+
+		srv, err := micro.AddService(nc, micro.Config{
+			Name:    "test-service-slow",
+			Version: "1.0.0",
+			ErrorHandler: func(svc micro.Service, natsErr *micro.NATSError) {
+				errCh <- natsErr
+			},
+		})
+		if err != nil {
+			t.Fatalf("Error adding service: %v", err)
+		}
+		defer srv.Stop()
+
+		msgLimit := 2
+		bytesLimit := -1
+
+		slowHandler := micro.HandlerFunc(func(req micro.Request) {
+			time.Sleep(100 * time.Millisecond)
+			req.Respond([]byte("OK"))
+		})
+
+		err = srv.AddEndpoint("slow", slowHandler,
+			micro.WithEndpointSubject("test.slow"),
+			micro.WithEndpointPendingLimits(msgLimit, bytesLimit))
+		if err != nil {
+			t.Fatalf("Error adding endpoint: %v", err)
+		}
+
+		// Publish 20 requests to force slow consumer
+		for i := range 20 {
+			if err := nc.Publish("test.slow", []byte("test")); err != nil {
+				t.Fatalf("Error publishing message %d: %v", i, err)
+			}
+		}
+
+		if err := nc.Flush(); err != nil {
+			t.Fatalf("Error flushing: %v", err)
+		}
+
+		select {
+		case err := <-errCh:
+			if !errors.Is(err, nats.ErrSlowConsumer) {
+				t.Fatalf("Expected ErrSlowConsumer but got: %v", err)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatal("Expected ErrSlowConsumer to be triggered within timeout")
+		}
+	})
+}
diff --git a/nats_test.go b/nats_test.go
index 8f4546a..d481948 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -1780,3 +1780,75 @@ func BenchmarkHeaderDecode(b *testing.B) {
 		})
 	}
 }
+
+// mockConn simulates a network connection that can fail and recover
+// after a number of attempts.
+type mockConn struct {
+	failures          int
+	temporaryFailures int
+	data              []byte
+}
+
+func (mc *mockConn) Write(p []byte) (int, error) {
+	if mc.failures < mc.temporaryFailures {
+		mc.failures++
+		return 0, &net.OpError{Op: "write", Net: "tcp", Err: errors.New("i/o timeout")}
+	}
+	mc.data = append(mc.data, p...)
+	return len(p), nil
+}
+
+func (mc *mockConn) SetWriteDeadline(t time.Time) error { return nil }
+func (mc *mockConn) Read(b []byte) (int, error)         { return 0, nil }
+func (mc *mockConn) Close() error                       { return nil }
+func (mc *mockConn) LocalAddr() net.Addr                { return nil }
+func (mc *mockConn) RemoteAddr() net.Addr               { return nil }
+func (mc *mockConn) SetDeadline(t time.Time) error      { return nil }
+func (mc *mockConn) SetReadDeadline(t time.Time) error  { return nil }
+
+func TestTimeoutWriterRecovery(t *testing.T) {
+	mc := &mockConn{temporaryFailures: 2}
+	tw := &timeoutWriter{
+		timeout: time.Second,
+		conn:    mc,
+	}
+	n, err := tw.Write([]byte("foo"))
+	if err == nil {
+		t.Fatal("Unexpected success")
+	}
+	if n != 0 {
+		t.Fatalf("Expected 0 bytes, got %d", n)
+	}
+	n, err = tw.Write([]byte("bar"))
+	if err == nil {
+		t.Fatal("Unexpected success")
+	}
+	if n != 0 {
+		t.Fatalf("Expected 0 bytes, got: %d", n)
+	}
+
+	// Should succeed since it was a temporary error.
+	testData := []byte("quux")
+	n, err = tw.Write(testData)
+	if err != nil {
+		t.Fatalf("Expected success, got: %v", err)
+	}
+	if n != len(testData) {
+		t.Fatalf("Expected %d, got: %d", len(testData), n)
+	}
+	if !bytes.Equal(mc.data, testData) {
+		t.Fatalf("Expected %q, got: %q", testData, mc.data)
+	}
+	testData2 := []byte("quuz")
+	n, err = tw.Write(testData2)
+	if err != nil {
+		t.Fatalf("Expected success, got: %v", err)
+	}
+	if n != len(testData2) {
+		t.Fatalf("Expected %d bytes written, got %d", len(testData2), n)
+	}
+	expectedData := append(testData, testData2...)
+	if !bytes.Equal(mc.data, expectedData) {
+		t.Fatalf("Expected data %q, got %q", expectedData, mc.data)
+	}
+}
diff --git a/test/kv_test.go b/test/kv_test.go
index 3fce540..0bcbce8 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -483,6 +483,54 @@ func TestKeyValueWatch(t *testing.T) {
 			t.Fatalf("Stop watcher did not return")
 		}
 	})
+
+	// Test channel-based error API integration with select patterns
+	t.Run("error channel with select", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH_ERROR"})
+		expectOk(t, err)
+
+		// Put some initial keys
+		_, err = kv.Put("test1", []byte("value1"))
+		expectOk(t, err)
+		_, err = kv.Put("test2", []byte("value2"))
+		expectOk(t, err)
+
+		watcher, err := kv.WatchAll()
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		updateCount := 0
+		var watchCompleted bool
+
+	Outer:
+		for !watchCompleted {
+			select {
+			case entry := <-watcher.Updates():
+				if entry == nil {
+					break Outer
+				}
+				updateCount++
+
+			case err := <-watcher.Error():
+				if err != nil {
+					t.Fatalf("Unexpected error from watcher error channel: %v", err)
+				}
+
+			case <-time.After(2 * time.Second):
+				t.Fatalf("Timeout waiting for watcher completion")
+			}
+		}
+
+		if updateCount < 2 {
+			t.Fatalf("Expected at least 2 updates, got %d", updateCount)
+		}
+	})
 }
 
 func TestKeyValueWatchContext(t *testing.T) {
@@ -885,6 +933,56 @@ func TestKeyValueListKeys(t *testing.T) {
 	if _, ok := kmap["age"]; !ok {
 		t.Fatalf("Expected %q to be only key present", "age")
 	}
+
+	// Test channel-based error API patterns
+	t.Run("error channel with select", func(t *testing.T) {
+		keys, err := kv.ListKeys()
+		expectOk(t, err)
+		defer keys.Stop()
+
+		var keyList []string
+		var completed bool
+
+		for !completed {
+			select {
+			case key, ok := <-keys.Keys():
+				if !ok {
+					completed = true
+					break
+				}
+				keyList = append(keyList, key)
+
+			case err := <-keys.Error():
+				if err != nil {
+					t.Fatalf("Unexpected error from error channel: %v", err)
+				}
+			}
+		}
+
+		if len(keyList) != 1 {
+			t.Fatalf("Expected 1 key using select pattern, got %d", len(keyList))
+		}
+	})
+
+	t.Run("error check after completion", func(t *testing.T) {
+		keys, err := kv.ListKeys()
+		expectOk(t, err)
+		defer keys.Stop()
+
+		var keyList []string
+		for key := range keys.Keys() {
+			keyList = append(keyList, key)
+		}
+
+		// Check for errors after completion - should not block and return nil
+		if err := <-keys.Error(); err != nil {
+			t.Fatalf("Unexpected error after completion: %v", err)
+		}
+
+		if len(keyList) != 1 {
+			t.Fatalf("Expected 1 key after completion check, got %d", len(keyList))
+		}
+	})
 }
 
 func TestKeyValueCrossAccounts(t *testing.T) {
diff --git a/test/nats_test.go b/test/nats_test.go
index 15bf492..873acc2 100644
--- a/test/nats_test.go
+++ b/test/nats_test.go
@@ -336,6 +336,28 @@ func TestUserJWTAndSeed(t *testing.T) {
 	nc.Close()
 }
 
+func TestUserCredentialBytes(t *testing.T) {
+	ts := runTrustServer()
+	defer ts.Shutdown()
+
+	// Test with JWT and seed as byte slices
+	jwtBytes := []byte(uJWT)
+	seedBytes := uSeed
+
+	nc, err := nats.Connect(ts.ClientURL(), nats.UserCredentialBytes(jwtBytes, seedBytes))
+	if err != nil {
+		t.Fatalf("Expected to connect with separate JWT and seed bytes, got %v", err)
+	}
+	nc.Close()
+
+	// Test with chained credentials (JWT and seed in same byte slice)
+	nc2, err := nats.Connect(ts.ClientURL(), nats.UserCredentialBytes([]byte(chained)))
+	if err != nil {
+		t.Fatalf("Expected to connect with chained credentials bytes, got %v", err)
+	}
+	nc2.Close()
+}
+
 // If we are using TLS and have multiple servers we try to match the IP
 // from a discovered server with the expected hostname for certs without IP
 // designations. In certain cases where there is a not authorized error and