[FIXED] Make sure on reverse match to compensate for wider target sub…

…jects. (#4032)

If we have a wider subject, meaning more tokens, but our subs end with a
wildcard token make sure that matches properly.

Signed-off-by: Derek Collison <derek@nats.io>

server/sublist.go

@@ -1,4 +1,4 @@
-// Copyright 2016-2020 The NATS Authors
+// Copyright 2016-2023 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
@@ -1529,13 +1529,13 @@ func (s *Sublist) ReverseMatch(subject string) *SublistResult {
 
 	result := &SublistResult{}
 
-	s.Lock()
+	s.RLock()
 	reverseMatchLevel(s.root, tokens, nil, result)
 	// Check for empty result.
 	if len(result.psubs) == 0 && len(result.qsubs) == 0 {
 		result = emptyResult
 	}
-	s.Unlock()
+	s.RUnlock()
 
 	return result
 }
@@ -1553,9 +1553,22 @@ func reverseMatchLevel(l *level, toks []string, n *node, results *SublistResult)
 				for _, n := range l.nodes {
 					reverseMatchLevel(n.next, toks[i+1:], n, results)
 				}
+				if l.pwc != nil {
+					reverseMatchLevel(l.pwc.next, toks[i+1:], n, results)
+				}
+				if l.fwc != nil {
+					getAllNodes(l, results)
+				}
 				return
 			}
 		}
+		// If the sub tree has a fwc at this position, match as well.
+		if l.fwc != nil {
+			getAllNodes(l, results)
+			return
+		} else if l.pwc != nil {
+			reverseMatchLevel(l.pwc.next, toks[i+1:], n, results)
+		}
 		n = l.nodes[t]
 		if n == nil {
 			break

server/sublist_test.go

@@ -1,4 +1,4 @@
-// Copyright 2016-2020 The NATS Authors
+// Copyright 2016-2023 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
@@ -1478,6 +1478,20 @@ func TestSublistReverseMatch(t *testing.T) {
 	verifyMember(r.psubs, fooBarBazSub, t)
 }
 
+func TestSublistReverseMatchWider(t *testing.T) {
+	s := NewSublistWithCache()
+	sub := newSub("uplink.*.*.>")
+	s.Insert(sub)
+
+	r := s.ReverseMatch("uplink.1.*.*.>")
+	verifyLen(r.psubs, 1, t)
+	verifyMember(r.psubs, sub, t)
+
+	r = s.ReverseMatch("uplink.1.2.3.>")
+	verifyLen(r.psubs, 1, t)
+	verifyMember(r.psubs, sub, t)
+}
+
 func TestSublistMatchWithEmptyTokens(t *testing.T) {
 	for _, test := range []struct {
 		name  string