fix(exporter): memory leak and lock contentions

cli/operator/node.go

@@ -538,13 +538,6 @@ func setupDB(logger *zap.Logger, eth2Network beaconprotocol.Network) (*kv.Badger
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to open db")
 	}
-	reopenDb := func() error {
-		if err := db.Close(); err != nil {
-			return errors.Wrap(err, "failed to close db")
-		}
-		db, err = kv.New(logger, cfg.DBOptions)
-		return errors.Wrap(err, "failed to reopen db")
-	}
 
 	migrationOpts := migrations.Options{
 		Db:      db,
@@ -561,24 +554,13 @@ func setupDB(logger *zap.Logger, eth2Network beaconprotocol.Network) (*kv.Badger
 
 	// If migrations were applied, we run a full garbage collection cycle
 	// to reclaim any space that may have been freed up.
-	// Close & reopen the database to trigger any unknown internal
-	// startup/shutdown procedures that the storage engine may have.
 	start := time.Now()
-	if err := reopenDb(); err != nil {
-		return nil, err
-	}
-
-	// Run a long garbage collection cycle with a timeout.
 	ctx, cancel := context.WithTimeout(context.Background(), 6*time.Minute)
 	defer cancel()
 	if err := db.FullGC(ctx); err != nil {
 		return nil, errors.Wrap(err, "failed to collect garbage")
 	}
 
-	// Close & reopen again.
-	if err := reopenDb(); err != nil {
-		return nil, err
-	}
 	logger.Debug("post-migrations garbage collection completed", fields.Duration(start))
 
 	return db, nil

ibft/storage/store.go

@@ -28,10 +28,9 @@
 // participantStorage struct
 // instanceType is what separates different iBFT eth2 duty types (attestation, proposal and aggregation)
 type participantStorage struct {
-	prefix         []byte
-	oldPrefix      string // kept back for cleanup
-	db             basedb.Database
-	participantsMu sync.Mutex
+	prefix    []byte
+	oldPrefix string // kept back for cleanup
+	db        basedb.Database
 }
 
 // New create new participant store
@@ -163,13 +162,7 @@
 }
 
 func (i *participantStorage) SaveParticipants(pk spectypes.ValidatorPK, slot phase0.Slot, newParticipants []spectypes.OperatorID) (updated bool, err error) {
-	i.participantsMu.Lock()
-	defer i.participantsMu.Unlock()
-
-	txn := i.db.Begin()
-	defer txn.Discard()
-
-	existingParticipants, err := i.getParticipants(txn, pk, slot)
+	existingParticipants, err := i.getParticipants(pk, slot)
 	if err != nil {
 		return false, fmt.Errorf("get participants %w", err)
 	}
@@ -179,14 +172,10 @@
 		return false, nil
 	}
 
-	if err := i.saveParticipants(txn, pk, slot, mergedParticipants); err != nil {
+	if err := i.saveParticipants(pk, slot, mergedParticipants); err != nil {
 		return false, fmt.Errorf("save participants: %w", err)
 	}
 
-	if err := txn.Commit(); err != nil {
-		return false, fmt.Errorf("commit transaction: %w", err)
-	}
-
 	return true, nil
 }
 
@@ -237,11 +226,11 @@
 }
 
 func (i *participantStorage) GetParticipants(pk spectypes.ValidatorPK, slot phase0.Slot) ([]spectypes.OperatorID, error) {
-	return i.getParticipants(nil, pk, slot)
+	return i.getParticipants(pk, slot)
 }
 
-func (i *participantStorage) getParticipants(txn basedb.ReadWriter, pk spectypes.ValidatorPK, slot phase0.Slot) ([]spectypes.OperatorID, error) {
-	val, found, err := i.get(txn, pk[:], slotToByteSlice(slot))
+func (i *participantStorage) getParticipants(pk spectypes.ValidatorPK, slot phase0.Slot) ([]spectypes.OperatorID, error) {
+	val, found, err := i.get(pk[:], slotToByteSlice(slot))
 	if err != nil {
 		return nil, err
 	}
@@ -253,12 +242,12 @@
 	return operators, nil
 }
 
-func (i *participantStorage) saveParticipants(txn basedb.ReadWriter, pk spectypes.ValidatorPK, slot phase0.Slot, operators []spectypes.OperatorID) error {
+func (i *participantStorage) saveParticipants(pk spectypes.ValidatorPK, slot phase0.Slot, operators []spectypes.OperatorID) error {
 	bytes, err := encodeOperators(operators)
 	if err != nil {
 		return fmt.Errorf("encode operators: %w", err)
 	}
-	if err := i.save(txn, bytes, pk[:], slotToByteSlice(slot)); err != nil {
+	if err := i.save(bytes, pk[:], slotToByteSlice(slot)); err != nil {
 		return fmt.Errorf("save to DB: %w", err)
 	}
 
@@ -271,14 +260,14 @@
 	return slices.Compact(allParticipants)
 }
 
-func (i *participantStorage) save(txn basedb.ReadWriter, value []byte, pk, slot []byte) error {
+func (i *participantStorage) save(value []byte, pk, slot []byte) error {
 	prefix := i.makePrefix(slot)
-	return i.db.Using(txn).Set(prefix, pk, value)
+	return i.db.Set(prefix, pk, value)
 }
 
-func (i *participantStorage) get(txn basedb.ReadWriter, pk, slot []byte) ([]byte, bool, error) {
+func (i *participantStorage) get(pk, slot []byte) ([]byte, bool, error) {
 	prefix := i.makePrefix(slot)
-	obj, found, err := i.db.Using(txn).Get(prefix, pk)
+	obj, found, err := i.db.Get(prefix, pk)
 	if err != nil {
 		return nil, false, err
 	}

message/validation/committee_info.go

@@ -0,0 +1,36 @@
+package validation
+
+import (
+	"github.com/attestantio/go-eth2-client/spec/phase0"
+	spectypes "github.com/ssvlabs/ssv-spec/types"
+)
+
+type CommitteeInfo struct {
+	committeeID      spectypes.CommitteeID
+	committee        []spectypes.OperatorID
+	signerIndices    map[spectypes.OperatorID]int
+	validatorIndices []phase0.ValidatorIndex
+}
+
+func newCommitteeInfo(
+	committeeID spectypes.CommitteeID,
+	operators []spectypes.OperatorID,
+	validatorIndices []phase0.ValidatorIndex,
+) CommitteeInfo {
+	signerIndices := make(map[spectypes.OperatorID]int)
+	for i, operator := range operators {
+		signerIndices[operator] = i
+	}
+
+	return CommitteeInfo{
+		committeeID:      committeeID,
+		committee:        operators,
+		signerIndices:    signerIndices,
+		validatorIndices: validatorIndices,
+	}
+}
+
+// keeping the method for readability and the comment
+func (ci *CommitteeInfo) signerIndex(signer spectypes.OperatorID) int {
+	return ci.signerIndices[signer] // existence must be checked by ErrSignerNotInCommittee
+}

message/validation/common_checks.go

@@ -40,7 +40,7 @@ func (mv *messageValidator) messageLateness(slot phase0.Slot, role spectypes.Run
 	case spectypes.RoleProposer, spectypes.RoleSyncCommitteeContribution:
 		ttl = 1 + LateSlotAllowance
 	case spectypes.RoleCommittee, spectypes.RoleAggregator:
-		ttl = phase0.Slot(mv.netCfg.Beacon.SlotsPerEpoch()) + LateSlotAllowance
+		ttl = MaxStoredSlots(mv.netCfg)
 	case spectypes.RoleValidatorRegistration, spectypes.RoleVoluntaryExit:
 		return 0
 	}

message/validation/consensus_state.go

@@ -1,39 +1,25 @@
 package validation
 
 import (
-	"sync"
-
 	"github.com/attestantio/go-eth2-client/spec/phase0"
-	spectypes "github.com/ssvlabs/ssv-spec/types"
 )
 
-// consensusID uniquely identifies a public key and role pair to keep track of state.
-type consensusID struct {
-	DutyExecutorID string
-	Role           spectypes.RunnerRole
-}
-
-// consensusState keeps track of the signers for a given public key and role.
-type consensusState struct {
-	state           map[spectypes.OperatorID]*OperatorState
+// ValidatorState keeps track of the signers for a given public key and role.
+type ValidatorState struct {
+	operators       []*OperatorState
 	storedSlotCount phase0.Slot
-	mu              sync.Mutex
 }
 
-func (cs *consensusState) GetOrCreate(signer spectypes.OperatorID) *OperatorState {
-	cs.mu.Lock()
-	defer cs.mu.Unlock()
-
-	if _, ok := cs.state[signer]; !ok {
-		cs.state[signer] = newOperatorState(cs.storedSlotCount)
+func (cs *ValidatorState) Signer(idx int) *OperatorState {
+	if cs.operators[idx] == nil {
+		cs.operators[idx] = newOperatorState(cs.storedSlotCount)
 	}
 
-	return cs.state[signer]
+	return cs.operators[idx]
 }
 
 type OperatorState struct {
-	mu              sync.RWMutex
-	state           []*SignerState // the slice index is slot % storedSlotCount
+	signers         []*SignerState // the slice index is slot % storedSlotCount
 	maxSlot         phase0.Slot
 	maxEpoch        phase0.Epoch
 	lastEpochDuties uint64
@@ -42,15 +28,12 @@ type OperatorState struct {
 
 func newOperatorState(size phase0.Slot) *OperatorState {
 	return &OperatorState{
-		state: make([]*SignerState, size),
+		signers: make([]*SignerState, size),
 	}
 }
 
 func (os *OperatorState) Get(slot phase0.Slot) *SignerState {
-	os.mu.RLock()
-	defer os.mu.RUnlock()
-
-	s := os.state[(uint64(slot) % uint64(len(os.state)))]
+	s := os.signers[(uint64(slot) % uint64(len(os.signers)))]
 	if s == nil || s.Slot != slot {
 		return nil
 	}
@@ -59,10 +42,7 @@ func (os *OperatorState) Get(slot phase0.Slot) *SignerState {
 }
 
 func (os *OperatorState) Set(slot phase0.Slot, epoch phase0.Epoch, state *SignerState) {
-	os.mu.Lock()
-	defer os.mu.Unlock()
-
-	os.state[uint64(slot)%uint64(len(os.state))] = state
+	os.signers[uint64(slot)%uint64(len(os.signers))] = state
 	if slot > os.maxSlot {
 		os.maxSlot = slot
 	}
@@ -76,16 +56,10 @@ func (os *OperatorState) Set(slot phase0.Slot, epoch phase0.Epoch, state *Signer
 }
 
 func (os *OperatorState) MaxSlot() phase0.Slot {
-	os.mu.RLock()
-	defer os.mu.RUnlock()
-
 	return os.maxSlot
 }
 
 func (os *OperatorState) DutyCount(epoch phase0.Epoch) uint64 {
-	os.mu.RLock()
-	defer os.mu.RUnlock()
-
 	if epoch == os.maxEpoch {
 		return os.lastEpochDuties
 	}

message/validation/consensus_state_test.go

@@ -12,7 +12,7 @@ func TestOperatorState(t *testing.T) {
 		size := phase0.Slot(10)
 		os := newOperatorState(size)
 		require.NotNil(t, os)
-		require.Equal(t, len(os.state), int(size))
+		require.Equal(t, len(os.signers), int(size))
 	})
 
 	t.Run("TestGetAndSet", func(t *testing.T) {
@@ -58,9 +58,9 @@ func TestOperatorState(t *testing.T) {
 
 		slot := phase0.Slot(5)
 		epoch := phase0.Epoch(1)
-		signerState := &SignerState{Slot: slot}
+		signerState1 := &SignerState{Slot: slot}
 
-		os.Set(slot, epoch, signerState)
+		os.Set(slot, epoch, signerState1)
 
 		require.Equal(t, os.DutyCount(epoch), uint64(1))
 		require.Equal(t, os.DutyCount(epoch-1), uint64(0))
@@ -82,9 +82,9 @@ func TestOperatorState(t *testing.T) {
 
 		slot := phase0.Slot(5)
 		epoch := phase0.Epoch(1)
-		signerState := &SignerState{Slot: slot}
+		signerState1 := &SignerState{Slot: slot}
 
-		os.Set(slot, epoch, signerState)
+		os.Set(slot, epoch, signerState1)
 		require.Equal(t, os.DutyCount(epoch), uint64(1))
 
 		slot2 := phase0.Slot(6)