From 68542ddc357f0eeaf27aed4b84aa1cf9954e4efd Mon Sep 17 00:00:00 2001
From: Nia Weiss <nia.weiss@molecula.com>
Date: Wed, 26 Aug 2020 09:40:23 -0400
Subject: [PATCH] force ranked cache recalculation in Top after a skipped
 invalidation

---
 cache.go      | 25 +++++++++++++++++++++++++
 cache_test.go | 29 +++++++++++++++++++++++++++++
 metrics.go    |  2 ++
 3 files changed, 56 insertions(+)

diff --git a/cache.go b/cache.go
index 9195f63c9..b1208f7ea 100644
--- a/cache.go
+++ b/cache.go
@@ -142,6 +142,7 @@ type rankCache struct {
 	entries      map[uint64]uint64
 	rankings     bitmapPairs // cached, ordered list
 	rankingsRead bool
+	dirty        bool
 
 	updateN    int
 	updateTime time.Time
@@ -173,6 +174,11 @@ func NewRankCache(maxEntries uint32) *rankCache {
 func (c *rankCache) Add(id uint64, n uint64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+
+	// Flag the cache as dirty.
+	// This forces recalculation if top is called before the cache is recalculated.
+	c.dirty = true
+
 	// Ignore if the column count is below the threshold,
 	// unless the count is 0, which is effectively used
 	// to clear the cache value.
@@ -190,6 +196,11 @@ func (c *rankCache) Add(id uint64, n uint64) {
 func (c *rankCache) BulkAdd(id uint64, n uint64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+
+	// Flag the cache as dirty.
+	// This forces recalculation if top is called before the cache is recalculated.
+	c.dirty = true
+
 	if n < c.thresholdValue {
 		delete(c.entries, id)
 		return
@@ -246,6 +257,11 @@ func (c *rankCache) invalidate() {
 	// Don't invalidate more than once every X seconds.
 	// TODO: consider making this configurable.
 	if time.Since(c.updateTime).Seconds() < 10 {
+		// Skipping recalculation means that the ranked cache's growth is unbounded.
+		// This is somewhat necessary for now since recalculation is not cheap.
+		// The cache will remain flagged as dirty and will be recalculated if Top is called.
+		// This may cause unexpected memory growth, so record it in metrics for debugging purposes.
+		c.stats.Count(MetricInvalidateCacheSkipped, 1, 1.0)
 		return
 	}
 	c.stats.Count(MetricInvalidateCache, 1, 1.0)
@@ -295,6 +311,9 @@ func (c *rankCache) recalculate() {
 			delete(c.entries, pair.ID)
 		}
 	}
+
+	// The cache is no longer dirty.
+	c.dirty = false
 }
 
 // SetStats defines the stats client used in the cache.
@@ -307,6 +326,12 @@ func (c *rankCache) Top() []bitmapPair {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
+	if c.dirty {
+		// The cache is dirty, so we need to recalculate it to get a consistent view.
+		c.stats.Count(MetricReadDirtyCache, 1, 1.0)
+		c.recalculate()
+	}
+
 	c.rankingsRead = true
 	return c.rankings
 }
diff --git a/cache_test.go b/cache_test.go
index d4c0982e9..fe7a3848a 100644
--- a/cache_test.go
+++ b/cache_test.go
@@ -15,6 +15,7 @@
 package pilosa_test
 
 import (
+	"reflect"
 	"testing"
 
 	"github.com/pilosa/pilosa/v2"
@@ -54,3 +55,31 @@ func TestCache_Rank_Threshold(t *testing.T) {
 		t.Fatalf("unexpected cache value after BulkAdd: %d!=%d expected\n", cache.Get(5), 0)
 	}
 }
+
+// Test that consecutive writes show up in Top.
+// On later writes, the cache skips recalculation to save CPU time.
+// This used to mean that the later writes would not show up in Top.
+// Now, the cache is flagged as dirty and recalculated during the call to Top.
+func TestCache_Rank_Dirty(t *testing.T) {
+	cacheSize := uint32(5)
+	cache := pilosa.NewRankCache(cacheSize)
+
+	type pair struct{ ID, Count uint64 }
+	expect := []pair{
+		{5, 2},
+		{4, 1},
+	}
+
+	for _, v := range expect {
+		cache.Add(v.ID, v.Count)
+	}
+
+	var got []pair
+	for _, p := range cache.Top() {
+		got = append(got, pair(p))
+	}
+
+	if !reflect.DeepEqual(expect, got) {
+		t.Fatalf("wrote %v but got %v", expect, got)
+	}
+}
diff --git a/metrics.go b/metrics.go
index 8d58f8a73..961af53c7 100644
--- a/metrics.go
+++ b/metrics.go
@@ -22,6 +22,8 @@ const (
 	MetricDeleteAvailableShard            = "delete_available_shard_total"
 	MetricRecalculateCache                = "recalculate_cache_total"
 	MetricInvalidateCache                 = "invalidate_cache_total"
+	MetricInvalidateCacheSkipped          = "invalidate_cache_skipped_total"
+	MetricReadDirtyCache                  = "dirty_cache_total"
 	MetricRankCacheLength                 = "rank_cache_length"
 	MetricCacheThresholdReached           = "cache_threshold_reached_total"
 	MetricRow                             = "query_row_total"