[FIXED] LeafNode: propagation of (no)interest issues

server/client.go

@@ -4866,19 +4866,32 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 						continue
 					}
 					// Remember that leaf in case we don't find any other candidate.
+					// We already start randomly in lqs slice, so we don't need
+					// to do a random swap if we already have an rsub like we do
+					// when src == ROUTER above.
 					if rsub == nil {
 						rsub = sub
 					}
 					continue
 				} else {
-					// We would be picking a route, but if we had remembered a "hub" leaf,
-					// then pick that one instead of the route.
-					if rsub != nil && rsub.client.kind == LEAF && rsub.client.isHubLeafNode() {
-						break
+					// We want to favor qsubs in our own cluster. If the routed
+					// qsub has an origin, it means that is on behalf of a leaf.
+					// We need to treat it differently.
+					if len(sub.origin) > 0 {
+						// If we already have an rsub, nothing to do. Also, do
+						// not pick a routed qsub for a LEAF origin cluster
+						// that is the same than where the message comes from.
+						if rsub == nil && (leafOrigin == _EMPTY_ || leafOrigin != bytesToString(sub.origin)) {
+							rsub = sub
+						}
+						continue
 					}
+					// This is a qsub that is local on the remote server (or
+					// we are connected to an older server and we don't know).
+					// Pick this one and be done.
 					rsub = sub
+					break
 				}
-				break
 			}
 
 			// Assume delivery subject is normal subject to this point.

server/const.go

@@ -174,6 +174,9 @@ const (
 	// MAX_HPUB_ARGS Maximum possible number of arguments from HPUB proto.
 	MAX_HPUB_ARGS = 4
 
+	// MAX_RSUB_ARGS Maximum possible number of arguments from a RS+/LS+ proto.
+	MAX_RSUB_ARGS = 6
+
 	// DEFAULT_MAX_CLOSED_CLIENTS is the maximum number of closed connections we hold onto.
 	DEFAULT_MAX_CLOSED_CLIENTS = 10000
 

server/gateway.go

@@ -1912,7 +1912,7 @@ func (c *client) processGatewayAccountSub(accName string) error {
 // the sublist if present.
 // <Invoked from outbound connection's readLoop>
 func (c *client) processGatewayRUnsub(arg []byte) error {
-	accName, subject, queue, err := c.parseUnsubProto(arg)
+	_, accName, subject, queue, err := c.parseUnsubProto(arg, true, false)
 	if err != nil {
 		return fmt.Errorf("processGatewaySubjectUnsub %s", err.Error())
 	}

server/leafnode.go

@@ -2353,6 +2353,42 @@ func keyFromSub(sub *subscription) string {
 	return sb.String()
 }
 
+const (
+	keyRoutedSub         = "R"
+	keyRoutedSubByte     = 'R'
+	keyRoutedLeafSub     = "L"
+	keyRoutedLeafSubByte = 'L'
+)
+
+// Helper function to build the key that prevents collisions between normal
+// routed subscriptions and routed subscriptions on behalf of a leafnode.
+// Keys will look like this:
+// "R foo"          -> plain routed sub on "foo"
+// "R foo bar"      -> queue routed sub on "foo", queue "bar"
+// "L foo bar"      -> plain routed leaf sub on "foo", leaf "bar"
+// "L foo bar baz"  -> queue routed sub on "foo", queue "bar", leaf "baz"
+func keyFromSubWithOrigin(sub *subscription) string {
+	var sb strings.Builder
+	sb.Grow(2 + len(sub.origin) + 1 + len(sub.subject) + 1 + len(sub.queue))
+	leaf := len(sub.origin) > 0
+	if leaf {
+		sb.WriteByte(keyRoutedLeafSubByte)
+	} else {
+		sb.WriteByte(keyRoutedSubByte)
+	}
+	sb.WriteByte(' ')
+	sb.Write(sub.subject)
+	if sub.queue != nil {
+		sb.WriteByte(' ')
+		sb.Write(sub.queue)
+	}
+	if leaf {
+		sb.WriteByte(' ')
+		sb.Write(sub.origin)
+	}
+	return sb.String()
+}
+
 // Lock should be held.
 func (c *client) writeLeafSub(w *bytes.Buffer, key string, n int32) {
 	if key == _EMPTY_ {
@@ -2403,12 +2439,21 @@ func (c *client) processLeafSub(argo []byte) (err error) {
 	args := splitArg(arg)
 	sub := &subscription{client: c}
 
+	delta := int32(1)
 	switch len(args) {
 	case 1:
 		sub.queue = nil
 	case 3:
 		sub.queue = args[1]
 		sub.qw = int32(parseSize(args[2]))
+		// TODO: (ik) We should have a non empty queue name and a queue
+		// weight >= 1. For 2.11, we may want to return an error if that
+		// is not the case, but for now just overwrite `delta` if queue
+		// weight is greater than 1 (it is possible after a reconnect/
+		// server restart to receive a queue weight > 1 for a new sub).
+		if sub.qw > 1 {
+			delta = sub.qw
+		}
 	default:
 		return fmt.Errorf("processLeafSub Parse Error: '%s'", arg)
 	}
@@ -2473,7 +2518,6 @@ func (c *client) processLeafSub(argo []byte) (err error) {
 	key := bytesToString(sub.sid)
 	osub := c.subs[key]
 	updateGWs := false
-	delta := int32(1)
 	if osub == nil {
 		c.subs[key] = sub
 		// Now place into the account sl.
@@ -2554,6 +2598,10 @@ func (c *client) processLeafUnsub(arg []byte) error {
 	// We store local subs by account and subject and optionally queue name.
 	// LS- will have the arg exactly as the key.
 	sub, ok := c.subs[string(arg)]
+	delta := int32(1)
+	if ok && len(sub.queue) > 0 {
+		delta = sub.qw
+	}
 	c.mu.Unlock()
 
 	if ok {
@@ -2563,14 +2611,14 @@ func (c *client) processLeafUnsub(arg []byte) error {
 
 	if !spoke {
 		// If we are routing subtract from the route map for the associated account.
-		srv.updateRouteSubscriptionMap(acc, sub, -1)
+		srv.updateRouteSubscriptionMap(acc, sub, -delta)
 		// Gateways
 		if updateGWs {
-			srv.gatewayUpdateSubInterest(acc.Name, sub, -1)
+			srv.gatewayUpdateSubInterest(acc.Name, sub, -delta)
 		}
 	}
 	// Now check on leafnode updates for other leaf nodes.
-	acc.updateLeafNodes(sub, -1)
+	acc.updateLeafNodes(sub, -delta)
 	return nil
 }