Repository: nats-io/nats.go
Pull request #1719 (2024-09-23 20:15:51)

[FIXED] Publish async not closing done and stall channels after failed retries
This fixes an issue in new JetStream API where if maximum number of retries
in `PublishMsgAsync` are reached and the publish failed, done and stall channels
are not closed.
Additionally, this fixes a potential race issue with modifying user msg when publishing.

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

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

coverage: 84.843% (-0.2%) from 84.998%
when pulling **de98a358fb2d30dd1cc385c0c9bde9f906622833 on fix-publish-async**
into **f0b1ea421c52eee17f89d528543b7affba18910b on main**.


Patch:
diff --git a/.github/workflows/dependencies.yaml b/.github/workflows/dependencies.yaml
new file mode 100644
index 0000000..5d9a7dc
--- /dev/null
+++ b/.github/workflows/dependencies.yaml
@@ -0,0 +1,61 @@
+name: License Check
+
+on:
+  push:
+    paths:
+      - 'go.mod'
+    branches:
+      - main
+
+jobs:
+  license-check:
+    runs-on: ubuntu-latest
+
+    env:
+      BRANCH_NAME: update-report-branch-${{ github.run_id }}
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0  # Fetch all history for all branches and tags
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.22'
+
+      - name: Install go-licenses
+        run: go install github.com/google/go-licenses@latest
+        # We need this step because of test dependencies and how they are handled in nats.go
+      - name: Run go mod tidy
+        run: go mod tidy
+      - name: Run license check
+        run: go-licenses report ./... --template dependencies.tpl > dependencies.md
+
+      - name: Configure git
+        run: |
+          git config user.name 'github-actions[bot]'
+          git config user.email 'github-actions[bot]@users.noreply.github.com'
+
+      - name: Check for changes
+        id: git_diff
+        run: |
+          git fetch
+          git diff --exit-code dependencies.md || echo "has_changes=true" >> $GITHUB_ENV
+
+      - name: Commit changes
+        if: env.has_changes == 'true'
+        run: |
+          git checkout -b "$BRANCH_NAME"
+          git add dependencies.md
+          git commit -m "Update dependencies.md"
+          git push -u origin "$BRANCH_NAME"
+
+      - name: Create Pull Request
+        if: env.has_changes == 'true'
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          gh pr create --title "Update dependencies.md" --body "This PR updates the dependencies report" --head "$BRANCH_NAME" --base main
+
diff --git a/dependencies.tpl b/dependencies.tpl
new file mode 100644
index 0000000..5d01380
--- /dev/null
+++ b/dependencies.tpl
@@ -0,0 +1,8 @@
+# External Dependencies
+
+This file lists the dependencies used in this repository.
+
+| Dependency                                       | License                                 |
+|--------------------------------------------------|-----------------------------------------|
+{{ range . }}| {{.Name}}                           | {{.LicenseName}}                        |
+{{ end }}
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index 36a44c8..3b94146 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -122,7 +122,8 @@ type (
 	// to operate on a stream.
 	StreamManager interface {
 		// CreateStream creates a new stream with given config and returns an
-		// interface to operate on it. If stream with given name already exists,
+		// interface to operate on it. If stream with given name already exists
+		// and its configuration differs from the provided one,
 		// ErrStreamNameAlreadyInUse is returned.
 		CreateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
 
diff --git a/jetstream/kv.go b/jetstream/kv.go
index ae892b4..91a1bdc 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -123,6 +123,7 @@ type (
 
 		// Update will update the value if the latest revision matches.
 		// If the provided revision is not the latest, Update will return an error.
+		// Update also resets the TTL associated with the key (if any).
 		Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)
 
 		// Delete will place a delete marker and leave all revisions. A history
diff --git a/jetstream/publish.go b/jetstream/publish.go
index d08cdf1..70e219a 100644
--- a/jetstream/publish.go
+++ b/jetstream/publish.go
@@ -81,6 +81,7 @@ type (
 		err        error
 		errCh      chan error
 		doneCh     chan *PubAck
+		reply      string
 	}
 
 	jetStreamClient struct {
@@ -280,17 +281,17 @@ func (js *jetStream) PublishMsgAsync(m *nats.Msg, opts ...PublishOpt) (PubAckFut
 	}
 
 	var id string
+	var reply string
 
 	// register new paf if not retrying
 	if paf == nil {
 		var err error
-		m.Reply, err = js.newAsyncReply()
-		defer func() { m.Reply = "" }()
+		reply, err = js.newAsyncReply()
 		if err != nil {
 			return nil, fmt.Errorf("nats: error creating async reply handler: %s", err)
 		}
-		id = m.Reply[js.replyPrefixLen:]
-		paf = &pubAckFuture{msg: m, jsClient: js.publisher, maxRetries: o.retryAttempts, retryWait: o.retryWait}
+		id = reply[js.replyPrefixLen:]
+		paf = &pubAckFuture{msg: m, jsClient: js.publisher, maxRetries: o.retryAttempts, retryWait: o.retryWait, reply: reply}
 		numPending, maxPending := js.registerPAF(id, paf)
 
 		if maxPending > 0 && numPending > maxPending {
@@ -303,10 +304,17 @@ func (js *jetStream) PublishMsgAsync(m *nats.Msg, opts ...PublishOpt) (PubAckFut
 		}
 	} else {
 		// when retrying, get the ID from existing reply subject
-		id = m.Reply[js.replyPrefixLen:]
+		reply = paf.reply
+		id = reply[js.replyPrefixLen:]
 	}
 
-	if err := js.conn.PublishMsg(m); err != nil {
+	pubMsg := &nats.Msg{
+		Subject: m.Subject,
+		Reply:   reply,
+		Data:    m.Data,
+		Header:  m.Header,
+	}
+	if err := js.conn.PublishMsg(pubMsg); err != nil {
 		js.clearPAF(id)
 		return nil, err
 	}
@@ -370,6 +378,31 @@ func (js *jetStream) handleAsyncReply(m *nats.Msg) {
 		return
 	}
 
+	closeStc := func() {
+		// Check on anyone stalled and waiting.
+		if js.publisher.stallCh != nil && len(js.publisher.acks) < js.publisher.maxpa {
+			close(js.publisher.stallCh)
+			js.publisher.stallCh = nil
+		}
+	}
+
+	closeDchFn := func() func() {
+		var dch chan struct{}
+		// Check on anyone one waiting on done status.
+		if js.publisher.doneCh != nil && len(js.publisher.acks) == 0 {
+			dch = js.publisher.doneCh
+			js.publisher.doneCh = nil
+		}
+		// Return function to close done channel which
+		// should be deferred so that error is processed and
+		// can be checked.
+		return func() {
+			if dch != nil {
+				close(dch)
+			}
+		}
+	}
+
 	doErr := func(err error) {
 		paf.err = err
 		if paf.errCh != nil {
@@ -378,7 +411,6 @@ func (js *jetStream) handleAsyncReply(m *nats.Msg) {
 		cb := js.publisher.asyncPublisherOpts.aecb
 		js.publisher.Unlock()
 		if cb != nil {
-			paf.msg.Reply = ""
 			cb(js, paf.msg, err)
 		}
 	}
@@ -387,7 +419,6 @@ func (js *jetStream) handleAsyncReply(m *nats.Msg) {
 	if len(m.Data) == 0 && m.Header.Get(statusHdr) == noResponders {
 		if paf.retries < paf.maxRetries {
 			paf.retries++
-			paf.msg.Reply = m.Subject
 			time.AfterFunc(paf.retryWait, func() {
 				js.publisher.Lock()
 				paf := js.getPAF(id)
@@ -408,25 +439,16 @@ func (js *jetStream) handleAsyncReply(m *nats.Msg) {
 			return
 		}
 		delete(js.publisher.acks, id)
+		closeStc()
+		defer closeDchFn()()
 		doErr(ErrNoStreamResponse)
 		return
 	}
 
 	// Remove
 	delete(js.publisher.acks, id)
-
-	// Check on anyone stalled and waiting.
-	if js.publisher.stallCh != nil && len(js.publisher.acks) < js.publisher.asyncPublisherOpts.maxpa {
-		close(js.publisher.stallCh)
-		js.publisher.stallCh = nil
-	}
-	// Check on anyone waiting on done status.
-	if js.publisher.doneCh != nil && len(js.publisher.acks) == 0 {
-		dch := js.publisher.doneCh
-		js.publisher.doneCh = nil
-		// Defer here so error is processed and can be checked.
-		defer close(dch)
-	}
+	closeStc()
+	defer closeDchFn()()
 
 	var pa pubAckResponse
 	if err := json.Unmarshal(m.Data, &pa); err != nil {
diff --git a/kv.go b/kv.go
index 4e7a3fd..3cade1f 100644
--- a/kv.go
+++ b/kv.go
@@ -54,6 +54,7 @@ type KeyValue interface {
 	// Create will add the key/value pair iff it does not exist.
 	Create(key string, value []byte) (revision uint64, err error)
 	// Update will update the value iff the latest revision matches.
+	// Update also resets the TTL associated with the key (if any).
 	Update(key string, value []byte, last uint64) (revision uint64, err error)
 	// Delete will place a delete marker and leave all revisions.
 	Delete(key string, opts ...DeleteOpt) error
diff --git a/micro/README.md b/micro/README.md
index e4ffe09..99949eb 100644
--- a/micro/README.md
+++ b/micro/README.md
@@ -206,35 +206,69 @@ Service IDs can be discovered by:
 ```sh
 nats req '$SRV.PING.EchoService' '' --replies=3
 
-8:59:41 Sending request on "$SRV.PING.EchoService"
-18:59:41 Received with rtt 688.042µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Sending request on "$SRV.PING.EchoService"
+13:03:04 Received with rtt 1.302208ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4K7","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 
-18:59:41 Received with rtt 704.167µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxvO","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Received with rtt 1.317ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4Kt","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 
-18:59:41 Received with rtt 707.875µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdy0a","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Received with rtt 1.320291ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4Lf","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 ```
 
 A specific service instance info can be retrieved:
 
 ```sh
-nats req '$SRV.INFO.EchoService.tNoopzL5Sp1M4qJZdhdxqC' ''
-
-19:40:06 Sending request on "$SRV.INFO.EchoService.tNoopzL5Sp1M4qJZdhdxqC"
-19:40:06 Received with rtt 282.375µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.info_response","description":"","subjects":["svc.echo"]}
+nats req '$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq
+
+13:04:19 Sending request on "$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
+13:04:19 Received with rtt 318.875µs
+{
+  "name": "EchoService",
+  "id": "x3Yuiq7g7MoxhXdxk7i4K7",
+  "version": "1.0.0",
+  "metadata": {},
+  "type": "io.nats.micro.v1.info_response",
+  "description": "",
+  "endpoints": [
+    {
+      "name": "default",
+      "subject": "svc.echo",
+      "queue_group": "q",
+      "metadata": null
+    }
+  ]
+}
 ```
 
 To get statistics for this service:
 
 ```sh
-nats req '$SRV.STATS.EchoService.tNoopzL5Sp1M4qJZdhdxqC' ''
-
-19:40:47 Sending request on "$SRV.STATS.EchoService.tNoopzL5Sp1M4qJZdhdxqC"
-19:40:47 Received with rtt 421.666µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.stats_response","started":"2023-05-22T16:59:39.938514Z","endpoints":[{"name":"default","subject":"svc.echo","metadata":null,"num_requests":0,"num_errors":0,"last_error":"","processing_time":0,"average_processing_time":0}]}
+nats req '$SRV.STATS.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq
+
+13:04:46 Sending request on "$SRV.STATS.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
+13:04:46 Received with rtt 678.25µs
+{
+  "name": "EchoService",
+  "id": "x3Yuiq7g7MoxhXdxk7i4K7",
+  "version": "1.0.0",
+  "metadata": {},
+  "type": "io.nats.micro.v1.stats_response",
+  "started": "2024-09-24T11:02:55.564771Z",
+  "endpoints": [
+    {
+      "name": "default",
+      "subject": "svc.echo",
+      "queue_group": "q",
+      "num_requests": 0,
+      "num_errors": 0,
+      "last_error": "",
+      "processing_time": 0,
+      "average_processing_time": 0
+    }
+  ]
+}
 ```
 
 ## Examples
diff --git a/nats.go b/nats.go
index 83638a5..3f12d61 100644
--- a/nats.go
+++ b/nats.go
@@ -86,6 +86,9 @@ const (
 
 	// MAX_CONNECTIONS_ERR is for when nats server denies the connection due to server max_connections limit
 	MAX_CONNECTIONS_ERR = "maximum connections exceeded"
+
+	// MAX_SUBSCRIPTIONS_ERR is for when nats server denies the connection due to server subscriptions limit
+	MAX_SUBSCRIPTIONS_ERR = "maximum subscriptions exceeded"
 )
 
 // Errors
@@ -141,6 +144,7 @@ var (
 	ErrNoResponders                = errors.New("nats: no responders available for request")
 	ErrMaxConnectionsExceeded      = errors.New("nats: server maximum connections exceeded")
 	ErrConnectionNotTLS            = errors.New("nats: connection is not tls")
+	ErrMaxSubscriptionsExceeded    = errors.New("nats: server maximum subscriptions exceeded")
 )
 
 // GetDefaultOptions returns default configuration options for the client.
@@ -2973,11 +2977,11 @@ func (nc *Conn) doReconnect(err error, forceReconnect bool) {
 
 // processOpErr handles errors from reading or parsing the protocol.
 // The lock should not be held entering this function.
-func (nc *Conn) processOpErr(err error) {
+func (nc *Conn) processOpErr(err error) bool {
 	nc.mu.Lock()
+	defer nc.mu.Unlock()
 	if nc.isConnecting() || nc.isClosed() || nc.isReconnecting() {
-		nc.mu.Unlock()
-		return
+		return false
 	}
 
 	if nc.Opts.AllowReconnect && nc.status == CONNECTED {
@@ -2997,14 +3001,12 @@ func (nc *Conn) processOpErr(err error) {
 		nc.clearPendingFlushCalls()
 
 		go nc.doReconnect(err, false)
-		nc.mu.Unlock()
-		return
+		return false
 	}
 
 	nc.changeConnStatus(DISCONNECTED)
 	nc.err = err
-	nc.mu.Unlock()
-	nc.close(CLOSED, true, nil)
+	return true
 }
 
 // dispatch is responsible for calling any async callbacks
@@ -3101,7 +3103,9 @@ func (nc *Conn) readLoop() {
 			err = nc.parse(buf)
 		}
 		if err != nil {
-			nc.processOpErr(err)
+			if shouldClose := nc.processOpErr(err); shouldClose {
+				nc.close(CLOSED, true, nil)
+			}
 			break
 		}
 	}
@@ -3431,15 +3435,17 @@ slowConsumer:
 	}
 }
 
-// processPermissionsViolation is called when the server signals a subject
-// permissions violation on either publish or subscribe.
-func (nc *Conn) processPermissionsViolation(err string) {
+// processTransientError is called when the server signals a non terminal error
+// which does not close the connection or trigger a reconnect.
+// This will trigger the async error callback if set.
+// These errors include the following:
+// - permissions violation on publish or subscribe
+// - maximum subscriptions exceeded
+func (nc *Conn) processTransientError(err error) {
 	nc.mu.Lock()
-	// create error here so we can pass it as a closure to the async cb dispatcher.
-	e := errors.New("nats: " + err)
-	nc.err = e
+	nc.err = err
 	if nc.Opts.AsyncErrorCB != nil {
-		nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, e) })
+		nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, err) })
 	}
 	nc.mu.Unlock()
 }
@@ -3671,15 +3677,17 @@ func (nc *Conn) processErr(ie string) {
 	// convert to lower case.
 	e := strings.ToLower(ne)
 
-	close := false
+	var close bool
 
 	// FIXME(dlc) - process Slow Consumer signals special.
 	if e == STALE_CONNECTION {
-		nc.processOpErr(ErrStaleConnection)
+		close = nc.processOpErr(ErrStaleConnection)
 	} else if e == MAX_CONNECTIONS_ERR {
-		nc.processOpErr(ErrMaxConnectionsExceeded)
+		close = nc.processOpErr(ErrMaxConnectionsExceeded)
 	} else if strings.HasPrefix(e, PERMISSIONS_ERR) {
-		nc.processPermissionsViolation(ne)
+		nc.processTransientError(fmt.Errorf("nats: %s", ne))
+	} else if strings.HasPrefix(e, MAX_SUBSCRIPTIONS_ERR) {
+		nc.processTransientError(ErrMaxSubscriptionsExceeded)
 	} else if authErr := checkAuthError(e); authErr != nil {
 		nc.mu.Lock()
 		close = nc.processAuthError(authErr)
@@ -5128,7 +5136,9 @@ func (nc *Conn) processPingTimer() {
 	nc.pout++
 	if nc.pout > nc.Opts.MaxPingsOut {
 		nc.mu.Unlock()
-		nc.processOpErr(ErrStaleConnection)
+		if shouldClose := nc.processOpErr(ErrStaleConnection); shouldClose {
+			nc.close(CLOSED, true, nil)
+		}
 		return
 	}
 


Test patch:
diff --git a/jetstream/test/publish_test.go b/jetstream/test/publish_test.go
index 2ab6de8..f79ad19 100644
--- a/jetstream/test/publish_test.go
+++ b/jetstream/test/publish_test.go
@@ -1453,6 +1453,7 @@ func TestPublishAsyncRetry(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
+			publishComplete := js.PublishAsyncComplete()
 			errs := make(chan error, 1)
 			go func() {
 				// create stream with delay so that publish will receive no responders
@@ -1476,6 +1477,12 @@ func TestPublishAsyncRetry(t *testing.T) {
 			case <-time.After(5 * time.Second):
 				t.Fatalf("Timeout waiting for ack")
 			}
+
+			select {
+			case <-publishComplete:
+			case <-time.After(5 * time.Second):
+				t.Fatalf("Did not receive completion signal")
+			}
 		})
 	}
 }
diff --git a/test/sub_test.go b/test/sub_test.go
index a27e142..559efc5 100644
--- a/test/sub_test.go
+++ b/test/sub_test.go
@@ -16,6 +16,7 @@ package test
 import (
 	"errors"
 	"fmt"
+	"os"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -1737,3 +1738,35 @@ func TestSubscriptionEvents(t *testing.T) {
 		close(blockChan)
 	})
 }
+
+func TestMaxSubscriptionsExceeded(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+	listen: 127.0.0.1:-1
+	max_subscriptions: 5
+	`))
+	defer os.Remove(conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	ch := make(chan error)
+	nc, err := nats.Connect(s.ClientURL(), nats.ErrorHandler(func(c *nats.Conn, s *nats.Subscription, err error) {
+		ch <- err
+	}))
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
+	}
+	defer nc.Close()
+
+	for i := 0; i < 6; i++ {
+		s, err := nc.Subscribe("foo", func(_ *nats.Msg) {})
+		if err != nil {
+			t.Fatalf("Error subscribing: %v", err)
+		}
+		defer s.Unsubscribe()
+	}
+
+	WaitOnChannel(t, ch, nats.ErrMaxSubscriptionsExceeded)
+
+	// wait for the server to process the SUBs
+	time.Sleep(100 * time.Millisecond)
+}