Fix committee test race (#12338)

* fix races in committee cache tests

* lint

* gratuitous defer ClearCache if ClearCache

* log warning to avoid failed block processing

* gaz

* add Clear to cache stubs

* fix Clear mistakes

* last fake cache fix

---------

Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>

beacon-chain/cache/BUILD.bazel

@@ -47,6 +47,7 @@ go_library(
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
         "@io_k8s_client_go//tools/cache:go_default_library",
         "@io_opencensus_go//trace:go_default_library",
     ],

beacon-chain/cache/active_balance.go

@@ -42,9 +42,16 @@ type BalanceCache struct {
 
 // NewEffectiveBalanceCache creates a new effective balance cache for storing/accessing total balance by epoch.
 func NewEffectiveBalanceCache() *BalanceCache {
-	return &BalanceCache{
-		cache: lruwrpr.New(maxBalanceCacheSize),
-	}
+	c := &BalanceCache{}
+	c.Clear()
+	return c
+}
+
+// Clear resets the SyncCommitteeCache to its initial state
+func (c *BalanceCache) Clear() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.cache = lruwrpr.New(maxBalanceCacheSize)
 }
 
 // AddTotalEffectiveBalance adds a new total effective balance entry for current balance for state `st` into the cache.

beacon-chain/cache/active_balance_disabled.go

@@ -3,16 +3,11 @@
 package cache
 
 import (
-	"sync"
-
-	lru "github.com/hashicorp/golang-lru"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/state"
 )
 
 // FakeBalanceCache is a fake struct with 1 LRU cache for looking up balance by epoch.
 type FakeBalanceCache struct {
-	cache *lru.Cache
-	lock  sync.RWMutex
 }
 
 // NewEffectiveBalanceCache creates a new effective balance cache for storing/accessing total balance by epoch.
@@ -29,3 +24,8 @@ func (c *FakeBalanceCache) AddTotalEffectiveBalance(st state.ReadOnlyBeaconState
 func (c *FakeBalanceCache) Get(st state.ReadOnlyBeaconState) (uint64, error) {
 	return 0, nil
 }
+
+// Clear is a stub.
+func (c *FakeBalanceCache) Clear() {
+	return
+}

beacon-chain/cache/committee.go

@@ -56,10 +56,17 @@ func committeeKeyFn(obj interface{}) (string, error) {
 
 // NewCommitteesCache creates a new committee cache for storing/accessing shuffled indices of a committee.
 func NewCommitteesCache() *CommitteeCache {
-	return &CommitteeCache{
-		CommitteeCache: lruwrpr.New(maxCommitteesCacheSize),
-		inProgress:     make(map[string]bool),
-	}
+	cc := &CommitteeCache{}
+	cc.Clear()
+	return cc
+}
+
+// Clear resets the CommitteeCache to its initial state
+func (c *CommitteeCache) Clear() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.CommitteeCache = lruwrpr.New(maxCommitteesCacheSize)
+	c.inProgress = make(map[string]bool)
 }
 
 // Committee fetches the shuffled indices by slot and committee index. Every list of indices

beacon-chain/cache/committee_disabled.go

@@ -69,3 +69,8 @@ func (c *FakeCommitteeCache) MarkInProgress(seed [32]byte) error {
 func (c *FakeCommitteeCache) MarkNotInProgress(seed [32]byte) error {
 	return nil
 }
+
+// Clear is a stub.
+func (c *FakeCommitteeCache) Clear() {
+	return
+}

beacon-chain/cache/proposer_indices.go

@@ -46,9 +46,16 @@ func proposerIndicesKeyFn(obj interface{}) (string, error) {
 
 // NewProposerIndicesCache creates a new proposer indices cache for storing/accessing proposer index assignments of an epoch.
 func NewProposerIndicesCache() *ProposerIndicesCache {
-	return &ProposerIndicesCache{
-		proposerIndicesCache: cache.NewFIFO(proposerIndicesKeyFn),
-	}
+	c := &ProposerIndicesCache{}
+	c.Clear()
+	return c
+}
+
+// Clear resets the ProposerIndicesCache to its initial state
+func (c *ProposerIndicesCache) Clear() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.proposerIndicesCache = cache.NewFIFO(proposerIndicesKeyFn)
 }
 
 // AddProposerIndices adds ProposerIndices object to the cache.

beacon-chain/cache/proposer_indices_disabled.go

@@ -33,3 +33,7 @@ func (c *FakeProposerIndicesCache) HasProposerIndices(r [32]byte) (bool, error)
 func (c *FakeProposerIndicesCache) Len() int {
 	return 0
 }
+
+// Clear is a stub.
+func (c *FakeProposerIndicesCache) Clear() {
+}

beacon-chain/cache/sync_committee.go

@@ -4,12 +4,14 @@ package cache
 
 import (
 	"sync"
+	"sync/atomic"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
+	log "github.com/sirupsen/logrus"
 	"k8s.io/client-go/tools/cache"
 )
 
@@ -31,8 +33,9 @@ var (
 // SyncCommitteeCache utilizes a FIFO cache to sufficiently cache validator position within sync committee.
 // It is thread safe with concurrent read write.
 type SyncCommitteeCache struct {
-	cache *cache.FIFO
-	lock  sync.RWMutex
+	cache   *cache.FIFO
+	lock    sync.RWMutex
+	cleared *atomic.Uint64
 }
 
 // Index position of all validators in sync committee where `currentSyncCommitteeRoot` is the
@@ -51,9 +54,17 @@ type positionInCommittee struct {
 
 // NewSyncCommittee initializes and returns a new SyncCommitteeCache.
 func NewSyncCommittee() *SyncCommitteeCache {
-	return &SyncCommitteeCache{
-		cache: cache.NewFIFO(keyFn),
-	}
+	c := &SyncCommitteeCache{cleared: &atomic.Uint64{}}
+	c.Clear()
+	return c
+}
+
+// Clear resets the SyncCommitteeCache to its initial state
+func (s *SyncCommitteeCache) Clear() {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.cleared.Add(1)
+	s.cache = cache.NewFIFO(keyFn)
 }
 
 // CurrentPeriodIndexPosition returns current period index position of a validator index with respect with
@@ -123,6 +134,10 @@ func (s *SyncCommitteeCache) idxPositionInCommittee(
 // current epoch and next epoch. This should be called when `current_sync_committee` and `next_sync_committee`
 // change and that happens every `EPOCHS_PER_SYNC_COMMITTEE_PERIOD`.
 func (s *SyncCommitteeCache) UpdatePositionsInCommittee(syncCommitteeBoundaryRoot [32]byte, st state.BeaconState) error {
+	// since we call UpdatePositionsInCommittee asynchronously, keep track of the cache value
+	// seen at the beginning of the routine and compare at the end before updating. If the underlying value has been
+	// cycled (new address), don't update it.
+	clearCount := s.cleared.Load()
 	csc, err := st.CurrentSyncCommittee()
 	if err != nil {
 		return err
@@ -162,6 +177,10 @@ func (s *SyncCommitteeCache) UpdatePositionsInCommittee(syncCommitteeBoundaryRoo
 
 	s.lock.Lock()
 	defer s.lock.Unlock()
+	if clearCount != s.cleared.Load() {
+		log.Warn("cache rotated during async committee update operation - abandoning cache update")
+		return nil
+	}
 
 	if err := s.cache.Add(&syncCommitteeIndexPosition{
 		currentSyncCommitteeRoot: syncCommitteeBoundaryRoot,

beacon-chain/cache/sync_committee_disabled.go

@@ -30,3 +30,8 @@ func (s *FakeSyncCommitteeCache) NextPeriodIndexPosition(root [32]byte, valIdx p
 func (s *FakeSyncCommitteeCache) UpdatePositionsInCommittee(syncCommitteeBoundaryRoot [32]byte, state state.BeaconState) error {
 	return nil
 }
+
+// Clear -- fake.
+func (s *FakeSyncCommitteeCache) Clear() {
+	return
+}

beacon-chain/core/helpers/BUILD.bazel

@@ -56,6 +56,7 @@ go_test(
         "weak_subjectivity_test.go",
     ],
     embed = [":go_default_library"],
+    race = "on",
     shard_count = 2,
     deps = [
         "//beacon-chain/cache:go_default_library",

beacon-chain/core/helpers/beacon_committee.go

@@ -382,10 +382,10 @@ func UpdateProposerIndicesInCache(ctx context.Context, state state.ReadOnlyBeaco
 
 // ClearCache clears the beacon committee cache and sync committee cache.
 func ClearCache() {
-	committeeCache = cache.NewCommitteesCache()
-	proposerIndicesCache = cache.NewProposerIndicesCache()
-	syncCommitteeCache = cache.NewSyncCommittee()
-	balanceCache = cache.NewEffectiveBalanceCache()
+	committeeCache.Clear()
+	proposerIndicesCache.Clear()
+	syncCommitteeCache.Clear()
+	balanceCache.Clear()
 }
 
 // computeCommittee returns the requested shuffled committee out of the total committees using

beacon-chain/core/helpers/beacon_committee_test.go

@@ -91,6 +91,7 @@ func TestVerifyBitfieldLength_OK(t *testing.T) {
 
 func TestCommitteeAssignments_CannotRetrieveFutureEpoch(t *testing.T) {
 	ClearCache()
+	defer ClearCache()
 	epoch := primitives.Epoch(1)
 	state, err := state_native.InitializeFromProtoPhase0(&ethpb.BeaconState{
 		Slot: 0, // Epoch 0.
@@ -101,6 +102,8 @@ func TestCommitteeAssignments_CannotRetrieveFutureEpoch(t *testing.T) {
 }
 
 func TestCommitteeAssignments_NoProposerForSlot0(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, 4*params.BeaconConfig().SlotsPerEpoch)
 	for i := 0; i < len(validators); i++ {
 		var activationEpoch primitives.Epoch
@@ -118,7 +121,6 @@ func TestCommitteeAssignments_NoProposerForSlot0(t *testing.T) {
 		RandaoMixes: make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector),
 	})
 	require.NoError(t, err)
-	ClearCache()
 	_, proposerIndexToSlots, err := CommitteeAssignments(context.Background(), state, 0)
 	require.NoError(t, err, "Failed to determine CommitteeAssignments")
 	for _, ss := range proposerIndexToSlots {
@@ -188,6 +190,7 @@ func TestCommitteeAssignments_CanRetrieve(t *testing.T) {
 		},
 	}
 
+	defer ClearCache()
 	for i, tt := range tests {
 		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
 			ClearCache()
@@ -255,6 +258,8 @@ func TestCommitteeAssignments_CannotRetrieveOlderThanSlotsPerHistoricalRoot(t *t
 }
 
 func TestCommitteeAssignments_EverySlotHasMin1Proposer(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	// Initialize test with 256 validators, each slot and each index gets 4 validators.
 	validators := make([]*ethpb.Validator, 4*params.BeaconConfig().SlotsPerEpoch)
 	for i := 0; i < len(validators); i++ {
@@ -269,7 +274,6 @@ func TestCommitteeAssignments_EverySlotHasMin1Proposer(t *testing.T) {
 		RandaoMixes: make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector),
 	})
 	require.NoError(t, err)
-	ClearCache()
 	epoch := primitives.Epoch(1)
 	_, proposerIndexToSlots, err := CommitteeAssignments(context.Background(), state, epoch)
 	require.NoError(t, err, "Failed to determine CommitteeAssignments")
@@ -376,6 +380,7 @@ func TestVerifyAttestationBitfieldLengths_OK(t *testing.T) {
 		},
 	}
 
+	defer ClearCache()
 	for i, tt := range tests {
 		ClearCache()
 		require.NoError(t, state.SetSlot(tt.stateSlot))
@@ -390,6 +395,7 @@ func TestVerifyAttestationBitfieldLengths_OK(t *testing.T) {
 
 func TestUpdateCommitteeCache_CanUpdate(t *testing.T) {
 	ClearCache()
+	defer ClearCache()
 	validatorCount := params.BeaconConfig().MinGenesisActiveValidatorCount
 	validators := make([]*ethpb.Validator, validatorCount)
 	indices := make([]primitives.ValidatorIndex, validatorCount)

beacon-chain/core/helpers/rewards_penalties_test.go

@@ -75,6 +75,8 @@ func TestTotalActiveBalance(t *testing.T) {
 }
 
 func TestTotalActiveBal_ReturnMin(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	tests := []struct {
 		vCount int
 	}{
@@ -96,6 +98,8 @@ func TestTotalActiveBal_ReturnMin(t *testing.T) {
 }
 
 func TestTotalActiveBalance_WithCache(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	tests := []struct {
 		vCount    int
 		wantCount int

beacon-chain/core/helpers/sync_committee.go

@@ -25,9 +25,7 @@ var (
 // along with the sync committee root.
 // 1. Checks if the public key exists in the sync committee cache
 // 2. If 1 fails, checks if the public key exists in the input current sync committee object
-func IsCurrentPeriodSyncCommittee(
-	st state.BeaconState, valIdx primitives.ValidatorIndex,
-) (bool, error) {
+func IsCurrentPeriodSyncCommittee(st state.BeaconState, valIdx primitives.ValidatorIndex) (bool, error) {
 	root, err := syncPeriodBoundaryRoot(st)
 	if err != nil {
 		return false, err

beacon-chain/core/helpers/sync_committee_test.go

@@ -17,6 +17,8 @@ import (
 )
 
 func TestIsCurrentEpochSyncCommittee_UsingCache(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -37,7 +39,6 @@ func TestIsCurrentEpochSyncCommittee_UsingCache(t *testing.T) {
 	require.NoError(t, state.SetCurrentSyncCommittee(syncCommittee))
 	require.NoError(t, state.SetNextSyncCommittee(syncCommittee))
 
-	ClearCache()
 	r := [32]byte{'a'}
 	require.NoError(t, err, syncCommitteeCache.UpdatePositionsInCommittee(r, state))
 
@@ -47,6 +48,8 @@ func TestIsCurrentEpochSyncCommittee_UsingCache(t *testing.T) {
 }
 
 func TestIsCurrentEpochSyncCommittee_UsingCommittee(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -73,6 +76,8 @@ func TestIsCurrentEpochSyncCommittee_UsingCommittee(t *testing.T) {
 }
 
 func TestIsCurrentEpochSyncCommittee_DoesNotExist(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -99,6 +104,8 @@ func TestIsCurrentEpochSyncCommittee_DoesNotExist(t *testing.T) {
 }
 
 func TestIsNextEpochSyncCommittee_UsingCache(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -119,7 +126,6 @@ func TestIsNextEpochSyncCommittee_UsingCache(t *testing.T) {
 	require.NoError(t, state.SetCurrentSyncCommittee(syncCommittee))
 	require.NoError(t, state.SetNextSyncCommittee(syncCommittee))
 
-	ClearCache()
 	r := [32]byte{'a'}
 	require.NoError(t, err, syncCommitteeCache.UpdatePositionsInCommittee(r, state))
 
@@ -181,6 +187,8 @@ func TestIsNextEpochSyncCommittee_DoesNotExist(t *testing.T) {
 }
 
 func TestCurrentEpochSyncSubcommitteeIndices_UsingCache(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -201,7 +209,6 @@ func TestCurrentEpochSyncSubcommitteeIndices_UsingCache(t *testing.T) {
 	require.NoError(t, state.SetCurrentSyncCommittee(syncCommittee))
 	require.NoError(t, state.SetNextSyncCommittee(syncCommittee))
 
-	ClearCache()
 	r := [32]byte{'a'}
 	require.NoError(t, err, syncCommitteeCache.UpdatePositionsInCommittee(r, state))
 
@@ -211,6 +218,8 @@ func TestCurrentEpochSyncSubcommitteeIndices_UsingCache(t *testing.T) {
 }
 
 func TestCurrentEpochSyncSubcommitteeIndices_UsingCommittee(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -230,7 +239,6 @@ func TestCurrentEpochSyncSubcommitteeIndices_UsingCommittee(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, state.SetCurrentSyncCommittee(syncCommittee))
 	require.NoError(t, state.SetNextSyncCommittee(syncCommittee))
-
 	root, err := syncPeriodBoundaryRoot(state)
 	require.NoError(t, err)
 
@@ -252,6 +260,7 @@ func TestCurrentEpochSyncSubcommitteeIndices_UsingCommittee(t *testing.T) {
 
 func TestCurrentEpochSyncSubcommitteeIndices_DoesNotExist(t *testing.T) {
 	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -278,6 +287,8 @@ func TestCurrentEpochSyncSubcommitteeIndices_DoesNotExist(t *testing.T) {
 }
 
 func TestNextEpochSyncSubcommitteeIndices_UsingCache(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -298,7 +309,6 @@ func TestNextEpochSyncSubcommitteeIndices_UsingCache(t *testing.T) {
 	require.NoError(t, state.SetCurrentSyncCommittee(syncCommittee))
 	require.NoError(t, state.SetNextSyncCommittee(syncCommittee))
 
-	ClearCache()
 	r := [32]byte{'a'}
 	require.NoError(t, err, syncCommitteeCache.UpdatePositionsInCommittee(r, state))
 
@@ -335,6 +345,7 @@ func TestNextEpochSyncSubcommitteeIndices_UsingCommittee(t *testing.T) {
 
 func TestNextEpochSyncSubcommitteeIndices_DoesNotExist(t *testing.T) {
 	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -387,6 +398,8 @@ func TestUpdateSyncCommitteeCache_BadRoot(t *testing.T) {
 }
 
 func TestIsCurrentEpochSyncCommittee_SameBlockRoot(t *testing.T) {
+	ClearCache()
+	defer ClearCache()
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
 		AggregatePubkey: bytesutil.PadTo([]byte{}, params.BeaconConfig().BLSPubkeyLength),
@@ -412,7 +425,6 @@ func TestIsCurrentEpochSyncCommittee_SameBlockRoot(t *testing.T) {
 	require.NoError(t, state.SetCurrentSyncCommittee(syncCommittee))
 	require.NoError(t, state.SetNextSyncCommittee(syncCommittee))
 
-	ClearCache()
 	comIdxs, err := CurrentPeriodSyncSubcommitteeIndices(state, 200)
 	require.NoError(t, err)
 

beacon-chain/core/helpers/validators_test.go

@@ -179,6 +179,7 @@ func TestIsSlashableValidator_OK(t *testing.T) {
 func TestBeaconProposerIndex_OK(t *testing.T) {
 	params.SetupTestConfigCleanup(t)
 	ClearCache()
+	defer ClearCache()
 	c := params.BeaconConfig()
 	c.MinGenesisActiveValidatorCount = 16384
 	params.OverrideBeaconConfig(c)
@@ -222,6 +223,7 @@ func TestBeaconProposerIndex_OK(t *testing.T) {
 		},
 	}
 
+	defer ClearCache()
 	for _, tt := range tests {
 		ClearCache()
 		require.NoError(t, state.SetSlot(tt.slot))
@@ -234,6 +236,7 @@ func TestBeaconProposerIndex_OK(t *testing.T) {
 func TestBeaconProposerIndex_BadState(t *testing.T) {
 	params.SetupTestConfigCleanup(t)
 	ClearCache()
+	defer ClearCache()
 	c := params.BeaconConfig()
 	c.MinGenesisActiveValidatorCount = 16384
 	params.OverrideBeaconConfig(c)
@@ -345,6 +348,7 @@ func TestChurnLimit_OK(t *testing.T) {
 		{validatorCount: 1000000, wantedChurn: 15 /* validatorCount/churnLimitQuotient */},
 		{validatorCount: 2000000, wantedChurn: 30 /* validatorCount/churnLimitQuotient */},
 	}
+	defer ClearCache()
 	for _, test := range tests {
 		ClearCache()
 
@@ -516,6 +520,7 @@ func TestActiveValidatorIndices(t *testing.T) {
 			want: []primitives.ValidatorIndex{0, 2, 3},
 		},
 	}
+	defer ClearCache()
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			s, err := state_native.InitializeFromProtoPhase0(tt.args.state)