Kan/clean up secondary result index (#1515)

* Remove from secondary index

* Add tests for caches

* add concurrency test

* Apply suggestions from code review

Co-authored-by: Simon Zhu <simon.zsiyan@gmail.com>

Co-authored-by: Leo Zhang (zhangchiqing) <zhangchiqing@gmail.com>
Co-authored-by: Simon Zhu <simon.zsiyan@gmail.com>

engine/consensus/approvals/approvals_lru_cache.go

@@ -24,6 +24,9 @@ func NewApprovalsLRUCache(limit uint) *LruCache {
 	lru, _ := simplelru.NewLRU(int(limit), func(key interface{}, value interface{}) {
 		approval := value.(*flow.ResultApproval)
 		delete(byResultID[approval.Body.ExecutionResultID], approval.Body.PartialID())
+		if len(byResultID[approval.Body.ExecutionResultID]) == 0 {
+			delete(byResultID, approval.Body.ExecutionResultID)
+		}
 	})
 	return &LruCache{
 		lru:        lru,

engine/consensus/approvals/approvals_lru_cache_test.go

@@ -0,0 +1,49 @@
+package approvals
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// TestApprovalsCache_Get_Put_All tests common use cases for approvals cache.
+func TestApprovalsLRUCacheSecondaryIndexPurge(t *testing.T) {
+	numElements := uint(10)
+	cache := NewApprovalsLRUCache(numElements)
+	approvals := make([]*flow.ResultApproval, 2*numElements)
+	for i := range approvals {
+		approval := unittest.ResultApprovalFixture()
+		approvals[i] = approval
+		cache.Put(approval)
+		require.Equal(t, approval, cache.Get(approval.Body.PartialID()))
+	}
+
+	// LRU kept the last 10 approvals
+	for i := 10; i < 20; i++ {
+		approval := approvals[i]
+		require.Equal(t, approval, cache.Get(approval.Body.PartialID()))
+	}
+	require.Len(t, cache.byResultID, int(numElements))
+}
+
+func TestApprovalsLRUCacheSecondaryIndexPurgeConcurrently(t *testing.T) {
+	numElements := 100
+	cache := NewApprovalsLRUCache(uint(numElements))
+
+	var wg sync.WaitGroup
+
+	for i := 0; i < 2*numElements; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			approval := unittest.ResultApprovalFixture()
+			cache.Put(approval)
+		}()
+	}
+	wg.Wait()
+	require.Len(t, cache.byResultID, int(numElements))
+}

engine/consensus/approvals/caches_test.go

@@ -11,26 +11,30 @@ import (
 
 // TestApprovalsCache_Get_Put_All tests common use cases for approvals cache.
 func TestApprovalsCache_Get_Put_All(t *testing.T) {
-	cache := NewApprovalsCache(10)
-	approvals := make([]*flow.ResultApproval, 0, 10)
+	numElements := uint(10)
+	cache := NewApprovalsCache(numElements)
+	approvals := make([]*flow.ResultApproval, numElements)
 	for i := range approvals {
 		approval := unittest.ResultApprovalFixture()
 		approvals[i] = approval
 		require.True(t, cache.Put(approval.Body.PartialID(), approval))
 		require.Equal(t, approval, cache.Get(approval.Body.PartialID()))
 	}
 	require.ElementsMatch(t, approvals, cache.All())
+	require.Len(t, cache.All(), int(numElements))
 }
 
 // TestIncorporatedResultsCache_Get_Put_All tests common use cases for incorporated results cache.
 func TestIncorporatedResultsCache_Get_Put_All(t *testing.T) {
-	cache := NewIncorporatedResultsCache(10)
-	results := make([]*flow.IncorporatedResult, 0, 10)
+	numElements := uint(10)
+	cache := NewIncorporatedResultsCache(numElements)
+	results := make([]*flow.IncorporatedResult, numElements)
 	for i := range results {
 		result := unittest.IncorporatedResult.Fixture()
 		results[i] = result
 		require.True(t, cache.Put(result.ID(), result))
 		require.Equal(t, result, cache.Get(result.ID()))
 	}
 	require.ElementsMatch(t, results, cache.All())
+	require.Equal(t, int(numElements), len(cache.All()))
 }