bugfix: findCache must be limit-aware

cache results depend on the limit.

idx/memory/find_cache.go

@@ -60,6 +60,11 @@ type FindCache struct {
 	sync.RWMutex
 }
 
+type cacheKey struct {
+	patt  string
+	limit int64
+}
+
 type CacheResult struct {
 	nodes []*Node
 	err   error
@@ -85,15 +90,16 @@ func NewFindCache(size, invalidateQueueSize, invalidateMaxSize int, invalidateMa
 	return fc
 }
 
-func (c *FindCache) Get(orgId uint32, pattern string) (CacheResult, bool) {
+func (c *FindCache) Get(orgId uint32, pattern string, limit int64) (CacheResult, bool) {
 	c.RLock()
 	cache, ok := c.cache[orgId]
 	c.RUnlock()
 	if !ok {
 		findCacheMiss.Inc()
 		return CacheResult{}, ok
 	}
-	nodes, ok := cache.Get(pattern)
+	key := cacheKey{pattern, limit}
+	nodes, ok := cache.Get(key)
 	if !ok {
 		findCacheMiss.Inc()
 		return CacheResult{}, ok
@@ -102,12 +108,17 @@ func (c *FindCache) Get(orgId uint32, pattern string) (CacheResult, bool) {
 	return nodes.(CacheResult), ok
 }
 
-func (c *FindCache) Add(orgId uint32, pattern string, nodes []*Node, e error) {
+func (c *FindCache) Add(orgId uint32, pattern string, limit int64, nodes []*Node, e error) {
 	c.RLock()
 	cache, ok := c.cache[orgId]
 	backoff := c.backoff
 	c.RUnlock()
 
+	key := cacheKey{
+		patt:  pattern,
+		limit: limit,
+	}
+
 	cr := CacheResult{
 		nodes: nodes,
 		err:   e,
@@ -133,7 +144,7 @@ func (c *FindCache) Add(orgId uint32, pattern string, nodes []*Node, e error) {
 		}
 		c.Unlock()
 	}
-	cache.Add(pattern, cr)
+	cache.Add(key, cr)
 }
 
 // Purge clears the cache for the specified orgId
@@ -289,7 +300,7 @@ func (c *FindCache) processInvalidateQueue() {
 			}
 
 			for _, k := range cache.Keys() {
-				matches, err := find((*Tree)(tree), k.(string), 0)
+				matches, err := find((*Tree)(tree), k.(cacheKey).patt, 0)
 				if err != nil {
 					log.Errorf("memory-idx: checking if cache key %q matches any of the %d invalid patterns resulted in error: %s", k, len(reqs), err)
 				}

idx/memory/find_cache_test.go

@@ -86,7 +86,7 @@ func TestFindCache(t *testing.T) {
 	Convey("when findCache is empty", t, func() {
 		c := NewFindCache(10, 5, 2, 100*time.Millisecond, time.Second*2)
 		Convey("0 results should be returned", func() {
-			result, ok := c.Get(1, "foo.bar.*")
+			result, ok := c.Get(1, "foo.bar.*", 0)
 			So(ok, ShouldBeFalse)
 			So(result.nodes, ShouldHaveLength, 0)
 		})
@@ -97,10 +97,10 @@ func TestFindCache(t *testing.T) {
 			results, err := find((*Tree)(tree), pattern, 0)
 			So(err, ShouldBeNil)
 			So(results, ShouldHaveLength, 1)
-			c.Add(1, "foo.bar.*", results, nil)
+			c.Add(1, "foo.bar.*", 0, results, nil)
 			So(c.cache[1].Len(), ShouldEqual, 1)
 			Convey("when getting cached pattern", func() {
-				result, ok := c.Get(1, "foo.bar.*")
+				result, ok := c.Get(1, "foo.bar.*", 0)
 				So(ok, ShouldBeTrue)
 				So(result.nodes, ShouldHaveLength, 1)
 				Convey("After invalidating path that matches pattern", func() {
@@ -114,25 +114,25 @@ func TestFindCache(t *testing.T) {
 				})
 			})
 			Convey("when findCache invalidation falls behind", func() {
-				c.Add(1, "foo.{a,b,c}*.*", results, nil)
-				c.Add(1, "foo.{a,b,e}*.*", results, nil)
-				c.Add(1, "foo.{a,b,f}*.*", results, nil)
+				c.Add(1, "foo.{a,b,c}*.*", 0, results, nil)
+				c.Add(1, "foo.{a,b,e}*.*", 0, results, nil)
+				c.Add(1, "foo.{a,b,f}*.*", 0, results, nil)
 				c.triggerBackoff()
 				c.InvalidateFor(1, "foo.baz.foo.a.b.c.d.e.f.g.h")
 
 				So(len(c.cache), ShouldEqual, 0)
 				Convey("when adding to cache in backoff", func() {
-					c.Add(1, "foo.*.*", results, nil)
+					c.Add(1, "foo.*.*", 0, results, nil)
 					So(len(c.cache), ShouldEqual, 0)
-					result, ok := c.Get(1, "foo.*.*")
+					result, ok := c.Get(1, "foo.*.*", 0)
 					So(ok, ShouldBeFalse)
 					So(result.nodes, ShouldHaveLength, 0)
 				})
 				Convey("when adding to cache after backoff time", func() {
 					time.Sleep(time.Millisecond * 2500)
-					c.Add(1, "foo.*.*", results, nil)
+					c.Add(1, "foo.*.*", 0, results, nil)
 					So(len(c.cache), ShouldEqual, 1)
-					result, ok := c.Get(1, "foo.*.*")
+					result, ok := c.Get(1, "foo.*.*", 0)
 					So(ok, ShouldBeTrue)
 					So(result.nodes, ShouldHaveLength, 1)
 				})

idx/memory/memory.go

@@ -1276,7 +1276,7 @@ func (m *UnpartitionedMemoryIdx) findMaybeCached(tree *Tree, orgId uint32, patte
 		return find(tree, pattern, limit)
 	}
 
-	cr, ok := m.findCache.Get(orgId, pattern)
+	cr, ok := m.findCache.Get(orgId, pattern, limit)
 	if ok {
 		return cr.nodes, cr.err
 	}
@@ -1290,7 +1290,7 @@ func (m *UnpartitionedMemoryIdx) findMaybeCached(tree *Tree, orgId uint32, patte
 			return nil, err
 		}
 	}
-	m.findCache.Add(orgId, pattern, matchedNodes, err)
+	m.findCache.Add(orgId, pattern, limit, matchedNodes, err)
 	return matchedNodes, err
 }