[fixed] sub issue across leaf node connections - bad sub ref count

[matthiashanel]

This was caused by not sending subs across leaf node connections in some
cases but sending unsub in all cases. This imbalance caused
subscriptions to go away too soon. (ref count was off)

Signed-off-by: Matthias Hanel mh@synadia.com

The test TestJetStreamClusterInterestRetention failed locally for me, but that happened even without changes in this PR

client.go  (5 usages found)
    closeConnection  (2 usages found) ---- added as part of PR
        4585 srv.updateRouteSubscriptionMap(acc, sub, -1) . 
        4606 srv.updateRouteSubscriptionMap(acc, esub.sub, -(esub.n))
    processSubEx  (1 usage found) ---- does not call with client.kind LEAF
        2437 srv.updateRouteSubscriptionMap(acc, sub, 1)
    processUnsub  (1 usage found) ---- does not call with client.kind LEAF
        2776 srv.updateRouteSubscriptionMap(acc, sub, -1)
    unsubscribe  (1 usage found) ---- does not call with client.kind LEAF
        2709 c.srv.updateRouteSubscriptionMap(nsub.im.acc, nsub, -1)
leafnode.go  (2 usages found)
    processLeafSub  (1 usage found)
        1610 srv.updateRouteSubscriptionMap(acc, sub, 1) ---- check already existed
    processLeafUnsub  (1 usage found)
        1667 srv.updateRouteSubscriptionMap(acc, sub, -1) ---- added as part of PR
server.go  (1 usage found)
    updateRemoteSubscription  (1 usage found)
        3368 s.updateRouteSubscriptionMap(acc, sub, delta) ---- see callers below

client.go  (3 usages found)
    addShadowSub  (1 usage found) ------- It is my understanding from the call that this won't be called for client.kind LEAF
        2562 c.srv.updateRemoteSubscription(im.acc, &nsub, 1)
    deliverMsg  (2 usages found) ------- the below invocations are only done for client.kind == CLIENT || client.kind == SYSTEM
        3015 defer srv.updateRemoteSubscription(client.acc, sub, -1)
        3023 srv.updateRemoteSubscription(client.acc, sub, -1)

[matthiashanel]

I also ran my test with these traces and was not able to see the sub/unsub on raft subjects.

diff --git a/server/leafnode.go b/server/leafnode.go
index 325a7933..12de9af6 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -1509,6 +1509,8 @@ func (c *client) processLeafSub(argo []byte) (err error) {
                return nil
        }

+       srv.Warnf("processLeafSub %q", argo)
+
        // Copy so we do not reference a potentially large buffer
        arg := make([]byte, len(argo))
        copy(arg, argo)
@@ -1644,6 +1646,8 @@ func (c *client) processLeafUnsub(arg []byte) error {
        acc := c.acc
        srv := c.srv

+       srv.Warnf("processLeafUnsub %q", arg)
+
        c.mu.Lock()
        if c.isClosed() {
                c.mu.Unlock()
diff --git a/server/route.go b/server/route.go
index cc6c0f19..c9339b62 100644
--- a/server/route.go
+++ b/server/route.go
@@ -954,6 +954,8 @@ func (c *client) processRemoteUnsub(arg []byte) (err error) {
                return nil
        }

+       srv.Warnf("processRemoteUnsub %q", arg)
+
        updateGWs := false
        // We store local subs by account and subject and optionally queue name.
        // RS- will have the arg exactly as the key.
@@ -989,6 +991,8 @@ func (c *client) processRemoteSub(argo []byte, hasOrigin bool) (err error) {
                return nil
        }

+       srv.Warnf("processRemoteSub %q", argo)
+
        // Copy so we do not reference a potentially large buffer
        arg := make([]byte, len(argo))
        copy(arg, argo)

[derekcollison]

LGTM

[kozlovic]

LGTM