Pull Consumer Updates

go.mod

@@ -7,10 +7,10 @@ require (
 	github.com/klauspost/compress v1.13.4
 	github.com/minio/highwayhash v1.0.1
 	github.com/nats-io/jwt/v2 v2.2.0
-	github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc
+	github.com/nats-io/nats.go v1.13.1-0.20220106142636-ebcdff697d3f
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
-	golang.org/x/crypto v0.0.0-20211202192323-5770296d904e
+	golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3
 	golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1
 	golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1
 )

go.sum

@@ -18,13 +18,17 @@ github.com/nats-io/jwt/v2 v2.2.0 h1:Yg/4WFK6vsqMudRg91eBb7Dh6XeVcDMPHycDE8CfltE=
 github.com/nats-io/jwt/v2 v2.2.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
 github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc h1:SHr4MUUZJ/fAC0uSm2OzWOJYsHpapmR86mpw7q1qPXU=
 github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats.go v1.13.1-0.20220106142636-ebcdff697d3f h1:WtlVcCLq7NOOskKidJ1rRZoXwOTs6RIJSl0R/gIPBMg=
+github.com/nats-io/nats.go v1.13.1-0.20220106142636-ebcdff697d3f/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=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20211202192323-5770296d904e h1:MUP6MR3rJ7Gk9LEia0LP2ytiH6MuCfs7qYz+47jGdD8=
 golang.org/x/crypto v0.0.0-20211202192323-5770296d904e/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3 h1:0es+/5331RGQPcXlMfP+WrnIIS6dNnNRe0WB02W0F4M=
+golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

server/accounts.go

@@ -1,4 +1,4 @@
-// Copyright 2018-2021 The NATS Authors
+// Copyright 2018-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
@@ -1242,9 +1242,7 @@ func (a *Account) sendReplyInterestLostTrackLatency(si *serviceImport) {
 		Error:  "Request Timeout",
 	}
 	a.mu.RLock()
-	rc := si.rc
-	share := si.share
-	ts := si.ts
+	rc, share, ts := si.rc, si.share, si.ts
 	sl.RequestHeader = si.trackingHdr
 	a.mu.RUnlock()
 	if rc != nil {
@@ -1257,9 +1255,7 @@ func (a *Account) sendReplyInterestLostTrackLatency(si *serviceImport) {
 func (a *Account) sendBackendErrorTrackingLatency(si *serviceImport, reason rsiReason) {
 	sl := &ServiceLatency{}
 	a.mu.RLock()
-	rc := si.rc
-	share := si.share
-	ts := si.ts
+	rc, share, ts := si.rc, si.share, si.ts
 	sl.RequestHeader = si.trackingHdr
 	a.mu.RUnlock()
 	if rc != nil {
@@ -1568,14 +1564,17 @@ func (a *Account) removeRespServiceImport(si *serviceImport, reason rsiReason) {
 	}
 
 	a.mu.Lock()
+	c := a.ic
 	delete(a.exports.responses, si.from)
-	dest := si.acc
-	to := si.to
-	tracking := si.tracking
-	rc := si.rc
+	dest, to, tracking, rc, didDeliver := si.acc, si.to, si.tracking, si.rc, si.didDeliver
 	a.mu.Unlock()
 
-	if tracking && rc != nil {
+	// If we have a sid make sure to unsub.
+	if len(si.sid) > 0 && c != nil {
+		c.processUnsub(si.sid)
+	}
+
+	if tracking && rc != nil && !didDeliver {
 		a.sendBackendErrorTrackingLatency(si, reason)
 	}
 
@@ -1714,12 +1713,17 @@ func (a *Account) checkForReverseEntry(reply string, si *serviceImport, checkInt
 			var trackingCleanup bool
 			var rsi *serviceImport
 			acc.mu.Lock()
+			c := acc.ic
 			if rsi = acc.exports.responses[sre.msub]; rsi != nil && !rsi.didDeliver {
 				delete(acc.exports.responses, rsi.from)
 				trackingCleanup = rsi.tracking && rsi.rc != nil
 			}
 			acc.mu.Unlock()
-
+			// If we are doing explicit subs for all responses (e.g. bound to leafnode)
+			// we will have a non-empty sid here.
+			if rsi != nil && len(rsi.sid) > 0 && c != nil {
+				c.processUnsub(rsi.sid)
+			}
 			if trackingCleanup {
 				acc.sendReplyInterestLostTrackLatency(rsi)
 			}
@@ -1842,18 +1846,26 @@ func (a *Account) internalClient() *client {
 
 // Internal account scoped subscriptions.
 func (a *Account) subscribeInternal(subject string, cb msgHandler) (*subscription, error) {
+	return a.subscribeInternalEx(subject, cb, false)
+}
+
+// Creates internal subscription for service import responses.
+func (a *Account) subscribeServiceImportResponse(subject string) (*subscription, error) {
+	return a.subscribeInternalEx(subject, a.processServiceImportResponse, true)
+}
+
+func (a *Account) subscribeInternalEx(subject string, cb msgHandler, ri bool) (*subscription, error) {
 	a.mu.Lock()
-	c := a.internalClient()
 	a.isid++
-	sid := strconv.FormatUint(a.isid, 10)
+	c, sid := a.internalClient(), strconv.FormatUint(a.isid, 10)
 	a.mu.Unlock()
 
 	// This will happen in parsing when the account has not been properly setup.
 	if c == nil {
 		return nil, fmt.Errorf("no internal account client")
 	}
 
-	return c.processSub([]byte(subject), nil, []byte(sid), cb, false)
+	return c.processSubEx([]byte(subject), nil, []byte(sid), cb, false, false, ri)
 }
 
 // This will add an account subscription that matches the "from" from a service import entry.
@@ -2062,29 +2074,18 @@ func (a *Account) processServiceImportResponse(sub *subscription, c *client, _ *
 	c.processServiceImport(si, a, msg)
 }
 
-// Will create a wildcard subscription to handle interest graph propagation for all
-// service replies.
-// Lock should not be held.
-func (a *Account) createRespWildcard() []byte {
-	a.mu.Lock()
+// Will create the response prefix for fast generation of responses.
+// A wildcard subscription may be used handle interest graph propagation
+// for all service replies, unless we are bound to a leafnode.
+// Lock should be held.
+func (a *Account) createRespWildcard() {
 	var b = [baseServerLen]byte{'_', 'R', '_', '.'}
 	rn := a.prand.Uint64()
 	for i, l := replyPrefixLen, rn; i < len(b); i++ {
 		b[i] = digits[l%base]
 		l /= base
 	}
 	a.siReply = append(b[:], '.')
-	pre := a.siReply
-	wcsub := append(a.siReply, '>')
-	c := a.internalClient()
-	a.isid++
-	sid := strconv.FormatUint(a.isid, 10)
-	a.mu.Unlock()
-
-	// Create subscription and internal callback for all the wildcard response subjects.
-	c.processSubEx(wcsub, nil, []byte(sid), a.processServiceImportResponse, false, false, true)
-
-	return pre
 }
 
 // Test whether this is a tracked reply.
@@ -2097,17 +2098,27 @@ func isTrackedReply(reply []byte) bool {
 // FIXME(dlc) - probably do not have to use rand here. about 25ns per.
 func (a *Account) newServiceReply(tracking bool) []byte {
 	a.mu.Lock()
-	s, replyPre := a.srv, a.siReply
+	s := a.srv
 	if a.prand == nil {
 		var h maphash.Hash
 		h.WriteString(nuid.Next())
 		a.prand = rand.New(rand.NewSource(int64(h.Sum64())))
 	}
 	rn := a.prand.Uint64()
+
+	// Check if we need to create the reply here.
+	var createdSiReply bool
+	if a.siReply == nil {
+		a.createRespWildcard()
+		createdSiReply = true
+	}
+	replyPre, isBoundToLeafnode := a.siReply, a.lds != _EMPTY_
 	a.mu.Unlock()
 
-	if replyPre == nil {
-		replyPre = a.createRespWildcard()
+	// If we created the siReply and we are not bound to a leafnode
+	// we need to do the wildcard subscription.
+	if createdSiReply && !isBoundToLeafnode {
+		a.subscribeServiceImportResponse(string(append(replyPre, '>')))
 	}
 
 	var b [replyLen]byte
@@ -2127,6 +2138,7 @@ func (a *Account) newServiceReply(tracking bool) []byte {
 		reply = append(reply, s.sys.shash...)
 		reply = append(reply, '.', 'T')
 	}
+
 	return reply
 }
 
@@ -2256,12 +2268,19 @@ func (a *Account) addRespServiceImport(dest *Account, to string, osi *serviceImp
 		si.tracking = true
 		si.trackingHdr = header
 	}
+	isBoundToLeafnode := a.lds != _EMPTY_
 	a.mu.Unlock()
 
-	// We do not do individual subscriptions here like we do on configured imports.
+	// We might not do individual subscriptions here like we do on configured imports.
+	// If we are bound to a leafnode we do explicit subscriptions for these.
 	// We have an internal callback for all responses inbound to this account and
 	// will process appropriately there. This does not pollute the sublist and the caches.
 
+	if isBoundToLeafnode {
+		sub, _ := a.subscribeServiceImportResponse(nrr)
+		si.sid = sub.sid
+	}
+
 	// We do add in the reverse map such that we can detect loss of interest and do proper
 	// cleanup of this si as interest goes away.
 	dest.addReverseRespMapEntry(a, to, nrr)