optimization: (eventsyncer) don't remove QBFT instances to speed up historical syncing (#1615)

AKorpusenko · y0sher · MatusKysel · web-flow · commit db74a7ddc8f5 · 2024-08-13T12:31:09.000+03:00
* optimize badger delete action

* try another optimization

* use dropprefix instead of delete all keys

* drop messages only when own share or fullnode

* removed usage of store.CleanAllInstances

* remove storageMap from eventhandler

---------

Co-authored-by: y0sher &lt;lyosher@gmail.com&gt;
Co-authored-by: Matus Kysel &lt;matus@ssvlabs.io&gt;
diff --git a/cli/operator/node.go b/cli/operator/node.go
@@ -372,7 +372,6 @@ var StartNodeCmd = &cobra.Command{
 			logger,
 			executionClient,
 			validatorCtrl,
-			storageMap,
 			metricsReporter,
 			networkConfig,
 			nodeStorage,
@@ -645,7 +644,6 @@ func setupEventHandling(
 	logger *zap.Logger,
 	executionClient *executionclient.ExecutionClient,
 	validatorCtrl validator.Controller,
-	storageMap *ibftstorage.QBFTStores,
 	metricsReporter metricsreporter.MetricsReporter,
 	networkConfig networkconfig.NetworkConfig,
 	nodeStorage operatorstorage.Storage,
@@ -669,7 +667,6 @@ func setupEventHandling(
 		operatorDecrypter,
 		keyManager,
 		cfg.SSVOptions.ValidatorOptions.Beacon,
-		storageMap,
 		eventhandler.WithFullNode(),
 		eventhandler.WithLogger(logger),
 		eventhandler.WithMetrics(metricsReporter),
diff --git a/eth/ethtest/utils_test.go b/eth/ethtest/utils_test.go
@@ -190,7 +190,6 @@ func setupEventHandler(
 			operator.privateKey,
 			keyManager,
 			bc,
-			storageMap,
 			eventhandler.WithFullNode(),
 			eventhandler.WithLogger(logger),
 		)
@@ -222,7 +221,6 @@ func setupEventHandler(
 		operator.privateKey,
 		keyManager,
 		bc,
-		storageMap,
 		eventhandler.WithFullNode(),
 		eventhandler.WithLogger(logger),
 	)
diff --git a/eth/eventhandler/event_handler.go b/eth/eventhandler/event_handler.go
@@ -5,10 +5,11 @@ package eventhandler
 import (
 	"errors"
 	"fmt"
-	"github.com/ssvlabs/ssv/ekm"
 	"math/big"
 	"time"
 
+	"github.com/ssvlabs/ssv/ekm"
+
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	ethcommon "github.com/ethereum/go-ethereum/common"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
@@ -19,7 +20,6 @@ import (
 	"github.com/ssvlabs/ssv/eth/eventparser"
 	"github.com/ssvlabs/ssv/eth/executionclient"
 	"github.com/ssvlabs/ssv/eth/localevents"
-	qbftstorage "github.com/ssvlabs/ssv/ibft/storage"
 	"github.com/ssvlabs/ssv/logging/fields"
 	"github.com/ssvlabs/ssv/networkconfig"
 	operatordatastore "github.com/ssvlabs/ssv/operator/datastore"
@@ -66,7 +66,6 @@ type EventHandler struct {
 	operatorDecrypter keys.OperatorDecrypter
 	keyManager        ekm.KeyManager
 	beacon            beaconprotocol.BeaconNode
-	storageMap        *qbftstorage.QBFTStores
 
 	fullNode bool
 	logger   *zap.Logger
@@ -82,7 +81,6 @@ func New(
 	operatorDecrypter keys.OperatorDecrypter,
 	keyManager ekm.KeyManager,
 	beacon beaconprotocol.BeaconNode,
-	storageMap *qbftstorage.QBFTStores,
 	opts ...Option,
 ) (*EventHandler, error) {
 	eh := &EventHandler{
@@ -94,7 +92,6 @@ func New(
 		operatorDecrypter: operatorDecrypter,
 		keyManager:        keyManager,
 		beacon:            beacon,
-		storageMap:        storageMap,
 		logger:            zap.NewNop(),
 		metrics:           nopMetrics{},
 	}
diff --git a/eth/eventhandler/event_handler_test.go b/eth/eventhandler/event_handler_test.go
@@ -1324,7 +1324,6 @@ func setupEventHandler(t *testing.T, ctx context.Context, logger *zap.Logger, ne
 			operator.privateKey,
 			keyManager,
 			bc,
-			storageMap,
 			WithFullNode(),
 			WithLogger(logger),
 			WithMetrics(nopMetrics{}),
@@ -1363,7 +1362,6 @@ func setupEventHandler(t *testing.T, ctx context.Context, logger *zap.Logger, ne
 		operator.privateKey,
 		keyManager,
 		bc,
-		storageMap,
 		WithFullNode(),
 		WithLogger(logger))
 	if err != nil {
diff --git a/eth/eventhandler/handlers.go b/eth/eventhandler/handlers.go
@@ -14,10 +14,8 @@ import (
 
 	"github.com/ssvlabs/ssv/ekm"
 	"github.com/ssvlabs/ssv/eth/contract"
-	"github.com/ssvlabs/ssv/exporter/convert"
 	"github.com/ssvlabs/ssv/logging/fields"
 	"github.com/ssvlabs/ssv/operator/duties"
-	qbftstorage "github.com/ssvlabs/ssv/protocol/v2/qbft/storage"
 	ssvtypes "github.com/ssvlabs/ssv/protocol/v2/types"
 	registrystorage "github.com/ssvlabs/ssv/registry/storage"
 	"github.com/ssvlabs/ssv/storage/basedb"
@@ -360,15 +358,6 @@ func (eh *EventHandler) handleValidatorRemoved(txn basedb.Txn, event *contract.C
 		return emptyPK, &MalformedEventError{Err: ErrShareBelongsToDifferentOwner}
 	}
 
-	removeDecidedMessages := func(role convert.RunnerRole, store qbftstorage.QBFTStore) error {
-		messageID := convert.NewMsgID(eh.networkConfig.DomainType(), share.ValidatorPubKey[:], role)
-		return store.CleanAllInstances(logger, messageID[:])
-	}
-	err := eh.storageMap.Each(removeDecidedMessages)
-	if err != nil {
-		return emptyPK, fmt.Errorf("could not clean all decided messages: %w", err)
-	}
-
 	if err := eh.nodeStorage.Shares().Delete(txn, share.ValidatorPubKey[:]); err != nil {
 		return emptyPK, fmt.Errorf("could not remove validator share: %w", err)
 	}
@@ -378,7 +367,7 @@ func (eh *EventHandler) handleValidatorRemoved(txn basedb.Txn, event *contract.C
 		logger = logger.With(zap.String("validator_pubkey", hex.EncodeToString(share.ValidatorPubKey[:])))
 	}
 	if isOperatorShare {
-		err = eh.keyManager.RemoveShare(hex.EncodeToString(share.SharePubKey))
+		err := eh.keyManager.RemoveShare(hex.EncodeToString(share.SharePubKey))
 		if err != nil {
 			return emptyPK, fmt.Errorf("could not remove share from ekm storage: %w", err)
 		}
diff --git a/eth/eventsyncer/event_syncer_test.go b/eth/eventsyncer/event_syncer_test.go
@@ -27,7 +27,6 @@ import (
 	"github.com/ssvlabs/ssv/eth/executionclient"
 	"github.com/ssvlabs/ssv/eth/simulator"
 	"github.com/ssvlabs/ssv/eth/simulator/simcontract"
-	ibftstorage "github.com/ssvlabs/ssv/ibft/storage"
 	"github.com/ssvlabs/ssv/networkconfig"
 	operatordatastore "github.com/ssvlabs/ssv/operator/datastore"
 	"github.com/ssvlabs/ssv/operator/keys"
@@ -152,7 +151,6 @@ func setupEventHandler(
 	operatorData *registrystorage.OperatorData,
 	privateKey keys.OperatorPrivateKey,
 ) *eventhandler.EventHandler {
-	storageMap := ibftstorage.NewStores()
 	operatorDataStore := operatordatastore.New(operatorData)
 	testNetworkConfig := networkconfig.TestNetwork
 
@@ -187,7 +185,6 @@ func setupEventHandler(
 		privateKey,
 		keyManager,
 		bc,
-		storageMap,
 		eventhandler.WithFullNode(),
 		eventhandler.WithLogger(logger))
 
diff --git a/go.mod b/go.mod
@@ -9,7 +9,7 @@ require (
 	github.com/btcsuite/btcd/btcec/v2 v2.3.2
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/cornelk/hashmap v1.0.8
-	github.com/dgraph-io/badger/v4 v4.1.0
+	github.com/dgraph-io/badger/v4 v4.2.0
 	github.com/dgraph-io/ristretto v0.1.1
 	github.com/ethereum/go-ethereum v1.13.15
 	github.com/ferranbt/fastssz v0.1.3
diff --git a/go.sum b/go.sum
@@ -139,6 +139,8 @@ github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0/go.mod h1:v57UDF4pDQJcEfFUCRop3
 github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4=
 github.com/dgraph-io/badger/v4 v4.1.0 h1:E38jc0f+RATYrycSUf9LMv/t47XAy+3CApyYSq4APOQ=
 github.com/dgraph-io/badger/v4 v4.1.0/go.mod h1:P50u28d39ibBRmIJuQC/NSdBOg46HnHw7al2SW5QRHg=
+github.com/dgraph-io/badger/v4 v4.2.0 h1:kJrlajbXXL9DFTNuhhu9yCx7JJa4qpYWxtE8BzuWsEs=
+github.com/dgraph-io/badger/v4 v4.2.0/go.mod h1:qfCqhPoWDFJRx1gp5QwwyGo8xk1lbHUxvK9nK0OGAak=
 github.com/dgraph-io/ristretto v0.1.1-0.20211108053508-297c39e6640f h1:NBGp2JpfMtXmanFWt6f3gEdBtnLO5LupRvm3w4TXrvs=
 github.com/dgraph-io/ristretto v0.1.1-0.20211108053508-297c39e6640f/go.mod h1:fux0lOrBhrVCJd3lcTHsIJhq1T2rokOu6v9Vcb3Q9ug=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
diff --git a/ibft/genesisstorage/store.go b/ibft/genesisstorage/store.go
@@ -140,7 +140,7 @@ func (i *ibftStorage) CleanAllInstances(logger *zap.Logger, msgID []byte) error
 	prefix := i.prefix
 	prefix = append(prefix, msgID[:]...)
 	prefix = append(prefix, []byte(instanceKey)...)
-	_, err := i.db.DeletePrefix(prefix)
+	err := i.db.DropPrefix(prefix)
 	if err != nil {
 		return errors.Wrap(err, "failed to remove decided")
 	}
diff --git a/ibft/storage/store.go b/ibft/storage/store.go
@@ -143,7 +143,7 @@ func (i *ibftStorage) CleanAllInstances(logger *zap.Logger, msgID []byte) error
 	prefix := i.prefix
 	prefix = append(prefix, msgID[:]...)
 	prefix = append(prefix, []byte(instanceKey)...)
-	_, err := i.db.DeletePrefix(prefix)
+	err := i.db.DropPrefix(prefix)
 	if err != nil {
 		return errors.Wrap(err, "failed to remove decided")
 	}
diff --git a/storage/basedb/storage.go b/storage/basedb/storage.go
@@ -53,7 +53,6 @@ type Database interface {
 
 	// TODO: consider moving these functions into Reader and ReadWriter interfaces?
 	CountPrefix(prefix []byte) (int64, error)
-	DeletePrefix(prefix []byte) (int, error)
 	DropPrefix(prefix []byte) error
 	Update(fn func(Txn) error) error
 	Close() error
diff --git a/storage/kv/badger.go b/storage/kv/badger.go
@@ -158,22 +158,6 @@ func (b *BadgerDB) Delete(prefix []byte, key []byte) error {
 	})
 }
 
-// DeletePrefix all items with this prefix
-func (b *BadgerDB) DeletePrefix(prefix []byte) (int, error) {
-	count := 0
-	err := b.db.Update(func(txn *badger.Txn) error {
-		rawKeys := b.listRawKeys(prefix, txn)
-		for _, k := range rawKeys {
-			if err := txn.Delete(k); err != nil {
-				return err
-			}
-			count++
-		}
-		return nil
-	})
-	return count, err
-}
-
 // GetAll returns all the items of a given collection
 func (b *BadgerDB) GetAll(prefix []byte, handler func(int, basedb.Obj) error) error {
 	// we got issues when reading more than 100 items with iterator (items get mixed up)
@@ -249,11 +233,13 @@ func (b *BadgerDB) listRawKeys(prefix []byte, txn *badger.Txn) [][]byte {
 
 	opt := badger.DefaultIteratorOptions
 	opt.Prefix = prefix
+	opt.PrefetchValues = false
+
 	it := txn.NewIterator(opt)
 	defer it.Close()
+
 	for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
-		item := it.Item()
-		keys = append(keys, item.KeyCopy(nil))
+		keys = append(keys, it.Item().KeyCopy(nil))
 	}
 
 	return keys
diff --git a/storage/kv/badger_test.go b/storage/kv/badger_test.go
@@ -87,9 +87,6 @@ func TestBadgerEndToEnd(t *testing.T) {
 	require.False(t, found)
 
 	require.NoError(t, db.DropPrefix([]byte("prefix2")))
-	deleted, err := db.DeletePrefix([]byte("prefix1"))
-	require.NoError(t, err)
-	require.Equal(t, 1, deleted)
 }
 
 func TestBadgerDb_GetAll(t *testing.T) {
@@ -194,7 +191,4 @@ func getAllTest(t *testing.T, n int, db basedb.Database) {
 		visited[string(item.Key)] = item.Value
 	}
 	require.Equal(t, n, len(visited))
-	count, err := db.DeletePrefix(prefix)
-	require.NoError(t, err)
-	require.Equal(t, n, count)
 }

Original file line number	Diff line number	Diff line change
`@@ -140,7 +140,7 @@ func (i ibftStorage) CleanAllInstances(logger zap.Logger, msgID []byte) error`
`140`	`140`	`prefix := i.prefix`
`141`	`141`	`prefix = append(prefix, msgID[:]...)`
`142`	`142`	`prefix = append(prefix, []byte(instanceKey)...)`
`143`		`- _, err := i.db.DeletePrefix(prefix)`
	`143`	`+ err := i.db.DropPrefix(prefix)`
`144`	`144`	`if err != nil {`
`145`	`145`	`return errors.Wrap(err, "failed to remove decided")`
`146`	`146`	`}`
Original file line number	Diff line number	Diff line change
`@@ -143,7 +143,7 @@ func (i ibftStorage) CleanAllInstances(logger zap.Logger, msgID []byte) error`
`143`	`143`	`prefix := i.prefix`
`144`	`144`	`prefix = append(prefix, msgID[:]...)`
`145`	`145`	`prefix = append(prefix, []byte(instanceKey)...)`
`146`		`- _, err := i.db.DeletePrefix(prefix)`
	`146`	`+ err := i.db.DropPrefix(prefix)`
`147`	`147`	`if err != nil {`
`148`	`148`	`return errors.Wrap(err, "failed to remove decided")`
`149`	`149`	`}`