Merge pull request #10336 from vegaprotocol/9992-eth-block-max

feat: add config option to limit the about of blocks worth of logs to…

CHANGELOG.md

@@ -35,6 +35,7 @@
 - [9983](https://github.com/vegaprotocol/vega/issues/9983) - Implement cap and discount for transfer fees.
 - [9980](https://github.com/vegaprotocol/vega/issues/9980) - Add teams statistics API.
 - [9257](https://github.com/vegaprotocol/vega/issues/9257) - Add games details API
+- [9992](https://github.com/vegaprotocol/vega/issues/9992) - Add configuration to control the number of blocks worth of Ethereum events to read.
 - [9260](https://github.com/vegaprotocol/vega/issues/9260) - Enhance rewards API for competitions
 - [10180](https://github.com/vegaprotocol/vega/issues/10180) - Additional candle intervals
 - [10218](https://github.com/vegaprotocol/vega/issues/10218) - Volume discount stats shows volumes even if party doesn't qualify for a discount tier.

core/evtforward/ethereum/config.go

@@ -24,18 +24,21 @@ import (
 
 const (
 	defaultDurationBetweenTwoRetry = 20 * time.Second
+	maxEthereumBlocks              = 10000 // chosen because one of the validators wanted to use quicknode, and this is their limit
 )
 
 type Config struct {
 	// Level specifies the logging level of the Ethereum implementation of the
 	// Event Forwarder.
 	Level                  encoding.LogLevel `long:"log-level"`
+	MaxEthereumBlocks      uint64            `long:"max-ethereum-blocks"`
 	PollEventRetryDuration encoding.Duration
 }
 
 func NewDefaultConfig() Config {
 	return Config{
 		Level:                  encoding.LogLevel{Level: logging.InfoLevel},
 		PollEventRetryDuration: encoding.Duration{Duration: defaultDurationBetweenTwoRetry},
+		MaxEthereumBlocks:      maxEthereumBlocks,
 	}
 }

core/evtforward/ethereum/engine.go

@@ -25,8 +25,7 @@ import (
 )
 
 const (
-	engineLogger      = "engine"
-	maxEthereumBlocks = 1000 // 3+ hour worth of blocks?
+	engineLogger = "engine"
 )
 
 //go:generate go run github.com/golang/mock/mockgen -destination mocks/forwarder_mock.go -package mocks code.vegaprotocol.io/vega/core/evtforward/ethereum Forwarder
@@ -136,11 +135,11 @@ func (e *Engine) Start() {
 	})
 }
 
-func issueFilteringRequest(from, to uint64) (ok bool, actualTo uint64) {
+func issueFilteringRequest(from, to, nBlocks uint64) (ok bool, actualTo uint64) {
 	if from > to {
 		return false, 0
 	}
-	return true, min(from+maxEthereumBlocks, to)
+	return true, min(from+nBlocks, to)
 }
 
 func min(a, b uint64) uint64 {
@@ -151,38 +150,43 @@ func min(a, b uint64) uint64 {
 }
 
 func (e *Engine) gatherEvents(ctx context.Context) {
+	nBlocks := e.cfg.MaxEthereumBlocks
 	currentHeight := e.filterer.CurrentHeight(ctx)
 
 	// Ensure we are not issuing a filtering request for non-existing block.
-	if ok, nextHeight := issueFilteringRequest(e.nextCollateralBlockNumber, currentHeight); ok {
+	if ok, nextHeight := issueFilteringRequest(e.nextCollateralBlockNumber, currentHeight, nBlocks); ok {
 		e.filterer.FilterCollateralEvents(ctx, e.nextCollateralBlockNumber, nextHeight, func(event *commandspb.ChainEvent) {
 			e.forwarder.ForwardFromSelf(event)
 		})
 		e.nextCollateralBlockNumber = nextHeight + 1
 	}
 
 	// Ensure we are not issuing a filtering request for non-existing block.
-	if e.shouldFilterStakingBridge && e.nextStakingBlockNumber <= currentHeight {
-		e.filterer.FilterStakingEvents(ctx, e.nextStakingBlockNumber, currentHeight, func(event *commandspb.ChainEvent) {
-			e.forwarder.ForwardFromSelf(event)
-		})
-		e.nextStakingBlockNumber = currentHeight + 1
+	if e.shouldFilterStakingBridge {
+		if ok, nextHeight := issueFilteringRequest(e.nextStakingBlockNumber, currentHeight, nBlocks); ok {
+			e.filterer.FilterStakingEvents(ctx, e.nextStakingBlockNumber, nextHeight, func(event *commandspb.ChainEvent) {
+				e.forwarder.ForwardFromSelf(event)
+			})
+			e.nextStakingBlockNumber = nextHeight + 1
+		}
 	}
 
 	// Ensure we are not issuing a filtering request for non-existing block.
-	if e.shouldFilterVestingBridge && e.nextVestingBlockNumber <= currentHeight {
-		e.filterer.FilterVestingEvents(ctx, e.nextVestingBlockNumber, currentHeight, func(event *commandspb.ChainEvent) {
-			e.forwarder.ForwardFromSelf(event)
-		})
-		e.nextVestingBlockNumber = currentHeight + 1
+	if e.shouldFilterVestingBridge {
+		if ok, nextHeight := issueFilteringRequest(e.nextVestingBlockNumber, currentHeight, nBlocks); ok {
+			e.filterer.FilterVestingEvents(ctx, e.nextVestingBlockNumber, nextHeight, func(event *commandspb.ChainEvent) {
+				e.forwarder.ForwardFromSelf(event)
+			})
+			e.nextVestingBlockNumber = nextHeight + 1
+		}
 	}
 
 	// Ensure we are not issuing a filtering request for non-existing block.
-	if e.nextMultiSigControlBlockNumber <= currentHeight {
-		e.filterer.FilterMultisigControlEvents(ctx, e.nextMultiSigControlBlockNumber, currentHeight, func(event *commandspb.ChainEvent) {
+	if ok, nextHeight := issueFilteringRequest(e.nextMultiSigControlBlockNumber, currentHeight, nBlocks); ok {
+		e.filterer.FilterMultisigControlEvents(ctx, e.nextMultiSigControlBlockNumber, nextHeight, func(event *commandspb.ChainEvent) {
 			e.forwarder.ForwardFromSelf(event)
 		})
-		e.nextMultiSigControlBlockNumber = currentHeight + 1
+		e.nextMultiSigControlBlockNumber = nextHeight + 1
 	}
 }