Run error callback in defer

Signed-off-by: Piotr Piotrowski <[email protected]>

jetstream/jetstream.go

@@ -1066,10 +1066,8 @@ func (js *jetStream) CleanupPublisher() {
 			paf.errCh <- paf.err
 		}
 		if errCb != nil {
-			// clear reply subject so that new one is created on republish
-			js.publisher.Unlock()
-			errCb(js, paf.msg, ErrJetStreamPublisherClosed)
-			js.publisher.Lock()
+			// call error handler after releasing the mutex to avoid contention
+			defer errCb(js, paf.msg, ErrJetStreamPublisherClosed)
 		}
 		delete(js.publisher.acks, id)
 	}

jetstream/publish.go

@@ -1,4 +1,4 @@
-// Copyright 2022-2023 The NATS Authors
+// Copyright 2022-2024 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
@@ -467,11 +467,7 @@ func (js *jetStream) resetPendingAcksOnReconnect() {
 				paf.errCh <- paf.err
 			}
 			if errCb != nil {
-				js.publisher.Unlock()
-				// clear reply subject so that new one is created on republish
-				paf.msg.Reply = ""
-				errCb(js, paf.msg, nats.ErrDisconnected)
-				js.publisher.Lock()
+				defer errCb(js, paf.msg, nats.ErrDisconnected)
 			}
 			delete(js.publisher.acks, id)
 		}

jetstream/test/jetstream_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022-2023 The NATS Authors
+// Copyright 2022-2024 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

js.go

@@ -1,4 +1,4 @@
-// Copyright 2020-2023 The NATS Authors
+// Copyright 2020-2024 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
@@ -732,10 +732,7 @@ func (js *js) resetPendingAcksOnReconnect() {
 				paf.errCh <- paf.err
 			}
 			if errCb != nil {
-				// clear reply subject so that new one is created on republish
-				js.mu.Unlock()
-				errCb(js, paf.msg, ErrDisconnected)
-				js.mu.Lock()
+				defer errCb(js, paf.msg, ErrDisconnected)
 			}
 			delete(js.pafs, id)
 		}
@@ -768,10 +765,7 @@ func (js *js) CleanupPublisher() {
 			paf.errCh <- paf.err
 		}
 		if errCb != nil {
-			// clear reply subject so that new one is created on republish
-			js.mu.Unlock()
-			errCb(js, paf.msg, ErrJetStreamPublisherClosed)
-			js.mu.Lock()
+			defer errCb(js, paf.msg, ErrJetStreamPublisherClosed)
 		}
 		delete(js.pafs, id)
 	}