ethevents: improve web3 failure management

- fix data race and make it race free (hopefully)
- control over the block number to detect frozen web3 endpoints
- clean a bit the code (remove all legacy stuff)
- manage web3 endpoints in a better way

If a web3 connection fails, then the next one is directly choosen (if
any), instead of keeping to reconnect with the one already failed.

TODO: apply the same mechanism on the signaling oracle

Signed-off-by: p4u <[email protected]>

Author: p4u

cmd/dvotenode/dvotenode.go

@@ -104,9 +104,6 @@ func newConfig() (*config.DvoteCfg, config.Error) {
 	// ethereum node
 	globalCfg.EthConfig.SigningKey = *flag.StringP("ethSigningKey", "k", "",
 		"signing private Key (if not specified the Ethereum keystore will be used)")
-	// ethereum events
-	globalCfg.EthEventConfig.SubscribeOnly = *flag.Bool("ethSubscribeOnly", true,
-		"only subscribe to new ethereum events (do not read past log)")
 	// ethereum web3
 	globalCfg.W3Config.ChainType = *flag.StringP("ethChain", "c", "goerli",
 		fmt.Sprintf("Ethereum blockchain to use: %s", ethchain.AvailableChains))
@@ -223,7 +220,6 @@ func newConfig() (*config.DvoteCfg, config.Error) {
 
 	// ethereum node
 	viper.BindPFlag("ethConfig.SigningKey", flag.Lookup("ethSigningKey"))
-	viper.BindPFlag("ethEventConfig.SubscribeOnly", flag.Lookup("ethSubscribeOnly"))
 
 	// ethereum web3
 	viper.BindPFlag("w3Config.ChainType", flag.Lookup("ethChain"))
@@ -579,20 +575,9 @@ func main() {
 			}
 			// Start ethereum events (if at least one web3 endpoint configured)
 			if len(w3uris) > 0 {
-				log.Infof("using %+v web3 endpoints", w3uris)
-
 				var evh []ethevents.EventHandler
 				evh = append(evh, ethevents.HandleVochainOracle)
 
-				var initBlock int64
-				if !globalCfg.EthEventConfig.SubscribeOnly {
-					chainSpecs, err := ethchain.SpecsFor(globalCfg.W3Config.ChainType)
-					if err != nil {
-						log.Fatal("cannot get chain block to start looking for events")
-					}
-					initBlock = chainSpecs.StartingBlock
-				}
-
 				whiteListedAddr := []string{}
 				for _, addr := range globalCfg.VochainConfig.EthereumWhiteListAddrs {
 					if addr == "[]" {
@@ -612,12 +597,9 @@ func main() {
 					context.Background(),
 					w3uris,
 					globalCfg.W3Config.ChainType,
-					initBlock,
-					cm,
 					signer,
 					vnode,
 					evh,
-					sc,
 					whiteListedAddr); err != nil {
 					log.Fatal(err)
 				}

config/config.go

@@ -14,8 +14,6 @@ type DvoteCfg struct {
 	Ipfs *IPFSCfg
 	// EthConfig ethereum client config options
 	EthConfig *EthCfg
-	// EthEventConfig ethereum even subscription config options
-	EthEventConfig *EthEventCfg
 	// API api config options
 	API *API
 	// Metrics config options
@@ -56,13 +54,12 @@ func (c *DvoteCfg) ValidMode() bool {
 // NewGatewayConfig initializes the fields in the gateway config stuct
 func NewConfig() *DvoteCfg {
 	return &DvoteCfg{
-		W3Config:       new(W3Cfg),
-		VochainConfig:  new(VochainCfg),
-		Ipfs:           new(IPFSCfg),
-		EthConfig:      new(EthCfg),
-		EthEventConfig: new(EthEventCfg),
-		API:            new(API),
-		Metrics:        new(MetricsCfg),
+		W3Config:      new(W3Cfg),
+		VochainConfig: new(VochainCfg),
+		Ipfs:          new(IPFSCfg),
+		EthConfig:     new(EthCfg),
+		API:           new(API),
+		Metrics:       new(MetricsCfg),
 	}
 }
 
@@ -120,13 +117,6 @@ type W3Cfg struct {
 	W3External []string
 }
 
-type EthEventCfg struct {
-	// CensusSync if true census sync will be enabled
-	CensusSync bool
-	// SubscribeOnly if true only new received events will be processed, otherwise all events of the current chain will be processed
-	SubscribeOnly bool
-}
-
 // VochainCfg includes all possible config params needed by the Vochain
 type VochainCfg struct {
 	// Chain is the network name to connect with

dockerfiles/dvotenode/env.example

@@ -27,8 +27,6 @@ DVOTE_DEV=True
 #DVOTE_ETHCONFIG_BOOTNODES=
 #DVOTE_ETHCONFIG_TRUSTEDPEERS=
 #DVOTE_ETHCONFIG_NOWAITSYNC=False
-#DVOTE_ETHEVENTCONFIG_CENSUSSYNC=True
-#DVOTE_ETHEVENTCONFIG_SUBSCRIBEONLY=False
 #DVOTE_W3CONFIG_ROUTE=/web3
 #DVOTE_W3CONFIG_ENABLED=False
 #DVOTE_W3CONFIG_RPCPORT=9091

ethereum/ethevents/events.go

@@ -14,12 +14,10 @@ import (
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	cttypes "github.com/tendermint/tendermint/rpc/core/types"
 	ttypes "github.com/tendermint/tendermint/types"
-	"go.vocdoni.io/dvote/census"
 	"go.vocdoni.io/dvote/crypto/ethereum"
 	"go.vocdoni.io/dvote/data"
 	"go.vocdoni.io/dvote/types"
 	"go.vocdoni.io/dvote/vochain"
-	"go.vocdoni.io/dvote/vochain/scrutinizer"
 	"go.vocdoni.io/proto/build/go/models"
 
 	"github.com/ethereum/go-ethereum/ethclient"
@@ -28,7 +26,7 @@ import (
 )
 
 const (
-	readBlocksPast = 100
+	readBlocksPast = 200
 )
 
 var blockConfirmThreshold = map[models.SourceNetworkId]time.Duration{
@@ -42,8 +40,6 @@ var blockConfirmThreshold = map[models.SourceNetworkId]time.Duration{
 type EthereumEvents struct {
 	// contracts handle
 	VotingHandle *ethereumhandler.EthereumHandler
-	// dial web3 addresses
-	DialAddrs []string
 	// list of handler functions that will be called on events
 	// TODO: add context on callbacks
 	// TODO: return errors on callbacks
@@ -52,20 +48,18 @@ type EthereumEvents struct {
 	Signer *ethereum.SignKeys
 	// VochainApp is a pointer to the Vochain BaseApplication allowing to call SendTx method
 	VochainApp *vochain.BaseApplication
-	// Census is the census manager service
-	Census CensusManager
 	// EventProcessor handles events pending to process
 	EventProcessor *EventProcessor
-	// Scrutinizer
-	Scrutinizer *scrutinizer.Scrutinizer
 	// EthereumWhiteListAddrs
 	EthereumWhiteListAddrs map[common.Address]bool
 	// ContractsAddress
 	ContractsAddress []common.Address
 	// ContractsInfo holds useful info for working with the desired contracts
 	ContractsInfo map[string]*ethereumhandler.EthereumContract
 	// EthereumLastKnownBlock keeps track of the latest Ethereum known block
-	EthereumLastKnownBlock atomic.Value
+	EthereumLastKnownBlock uint64
+	// block confirm threshold for the network
+	blockConfirmThreshold time.Duration
 }
 
 type timedEvent struct {
@@ -105,8 +99,6 @@ func NewEthEvents(
 	contracts map[string]*ethereumhandler.EthereumContract,
 	srcNetworkId models.SourceNetworkId,
 	signer *ethereum.SignKeys,
-	w3Endpoints []string,
-	cens *census.Manager,
 	vocapp *vochain.BaseApplication,
 	ethereumWhiteList []string,
 ) (*EthereumEvents, error) {
@@ -136,8 +128,6 @@ func NewEthEvents(
 		srcNetworkId, confirmThreshold)
 	ethev := &EthereumEvents{
 		Signer:     signer,
-		DialAddrs:  w3Endpoints,
-		Census:     cens,
 		VochainApp: vocapp,
 		EventProcessor: &EventProcessor{
 			Events:                make(chan ethtypes.Log),
@@ -147,6 +137,7 @@ func NewEthEvents(
 		EthereumWhiteListAddrs: secureAddrList,
 		ContractsAddress:       contractsAddress,
 		ContractsInfo:          contracts,
+		blockConfirmThreshold:  confirmThreshold,
 	}
 
 	return ethev, nil
@@ -167,29 +158,24 @@ func (ev *EthereumEvents) SubscribeEthereumEventLogs(ctx context.Context) {
 	tctx, cancel := context.WithTimeout(ctx, types.EthereumReadTimeout)
 	defer cancel()
 	var err error
-	uLastBlockNumber, err := ev.VotingHandle.EthereumClient.BlockNumber(tctx)
+	lastBlockNumber, err := ev.VotingHandle.EthereumClient.BlockNumber(tctx)
 	if err != nil {
 		log.Fatalf("cannot get last block number: (%v)", err)
 	}
-	lastBlockNumber := int64(uLastBlockNumber)
-	fromBlock := ev.EthereumLastKnownBlock.Load().(int64)
-	if fromBlock == 0 {
-		ev.EthereumLastKnownBlock.Store(lastBlockNumber)
-		fromBlock = lastBlockNumber
-	}
+	atomic.StoreUint64(&ev.EthereumLastKnownBlock, lastBlockNumber)
 
 	// For security, even if subscribe only, force to process
 	// from some past blocks defined by `readBlocksPast`
-	if err := ev.processEventLogsFromTo(ctx, fromBlock-readBlocksPast,
+	if err := ev.processEventLogsFromTo(ctx, lastBlockNumber-readBlocksPast,
 		lastBlockNumber, ev.VotingHandle.EthereumClient); err != nil {
 		log.Errorf("cannot process event logs: %v", err)
 	}
 
 	// Subscribing from latest known block
-	log.Infof("subscribing to Ethereum Events from block %d", fromBlock)
+	log.Infof("subscribing to Ethereum Events from block %d", lastBlockNumber)
 	query := eth.FilterQuery{
 		Addresses: ev.ContractsAddress,
-		FromBlock: big.NewInt(fromBlock),
+		FromBlock: new(big.Int).SetUint64(lastBlockNumber),
 	}
 	logs := make(chan ethtypes.Log, 30) // give it some buffer as recommended by the package library
 	var sub eth.Subscription
@@ -204,38 +190,44 @@ func (ev *EthereumEvents) SubscribeEthereumEventLogs(ctx context.Context) {
 	}
 
 	// Monitorize ethereum client connection
-	connFailure := make(chan bool, 1)
-	go func(chan bool) {
+	connFailure := make(chan error, 1)
+	go func(chan error) {
+		time.Sleep(ev.blockConfirmThreshold)
 		for {
-			time.Sleep(time.Second * 5)
-			tctx, cancel := context.WithTimeout(ctx, types.EthereumReadTimeout)
-			defer cancel()
-			block, err := ev.VotingHandle.EthereumClient.BlockNumber(tctx)
-			if err != nil {
-				log.Warnf("cannot check ethereum connection while running the event processor, %s", err)
-				connFailure <- false
+			select {
+			case <-ctx.Done():
 				return
+			default:
+				tctx, cancel := context.WithTimeout(ctx, types.EthereumReadTimeout)
+				block, err := ev.VotingHandle.EthereumClient.BlockNumber(tctx)
+				cancel()
+				if err != nil {
+					connFailure <- err
+					return
+				}
+				if atomic.LoadUint64(&ev.EthereumLastKnownBlock) >= block {
+					connFailure <- fmt.Errorf("web3 frozen, block have not changed on the last %s",
+						ev.blockConfirmThreshold)
+					return
+				}
+				atomic.StoreUint64(&ev.EthereumLastKnownBlock, block)
+				time.Sleep(ev.blockConfirmThreshold)
 			}
-			ev.EthereumLastKnownBlock.Store(int64(block))
 		}
 	}(connFailure)
 
 	// Block forever and start processing events
 	for {
 		select {
-		case <-sub.Err():
+		case err := <-sub.Err():
 			ev.EventProcessor.eventProcessorRunning = false
-			ev.VotingHandle = nil
-			log.Warn("ethereum events subscription error on channel")
-			log.Infof("restarting ethereum events subscription with web3: %+v", ev.DialAddrs)
+			log.Warnf("ethereum events connection error: %v", err)
 			return
 		case event := <-logs:
 			ev.EventProcessor.Events <- event
-		case <-connFailure:
+		case err := <-connFailure:
 			ev.EventProcessor.eventProcessorRunning = false
-			ev.VotingHandle = nil
-			log.Warn("ethereum events subscription error on channel")
-			log.Infof("restarting ethereum events subscription with web3: %+v", ev.DialAddrs)
+			log.Warnf("ethereum events connection failure: %v", err)
 			return
 		}
 	}
@@ -244,7 +236,7 @@ func (ev *EthereumEvents) SubscribeEthereumEventLogs(ctx context.Context) {
 // ReadEthereumEventLogs reads the oracle
 // defined smart contract and looks for events.
 func (ev *EthereumEvents) processEventLogsFromTo(ctx context.Context,
-	from, to int64, client *ethclient.Client) error {
+	from, to uint64, client *ethclient.Client) error {
 	for name, contract := range ev.ContractsInfo {
 		if !contract.ListenForEvents {
 			continue
@@ -254,8 +246,8 @@ func (ev *EthereumEvents) processEventLogsFromTo(ctx context.Context,
 	}
 	log.Infof("reading ethereum events from block %d to %d", from, to)
 	query := eth.FilterQuery{
-		FromBlock: big.NewInt(from),
-		ToBlock:   big.NewInt(to),
+		FromBlock: new(big.Int).SetUint64(from),
+		ToBlock:   new(big.Int).SetUint64(to),
 		Addresses: ev.ContractsAddress,
 	}
 

ethereum/ethevents/handlers.go

@@ -71,7 +71,7 @@ var ethereumEventList = map[string]string{
 
 // Number of blocks to add to the process start block to start as soon as possible the process.
 // This start block addition takes into account that can be some delay on the Vochain commit operation.
-const ProcessStartBlockDelay = 10
+const processStartBlockDelay = 10
 
 // HandleVochainOracle handles the events on ethereum for the Oracle.
 func HandleVochainOracle(ctx context.Context, event *ethtypes.Log, e *EthereumEvents) error {
@@ -133,7 +133,7 @@ func HandleVochainOracle(ctx context.Context, event *ethtypes.Log, e *EthereumEv
 		if processTx.Process.Status == models.ProcessStatus_READY &&
 			processTx.Process.StartBlock == 1 &&
 			processTx.Process.Mode.AutoStart {
-			processTx.Process.StartBlock = e.VochainApp.Height() + ProcessStartBlockDelay
+			processTx.Process.StartBlock = e.VochainApp.Height() + processStartBlockDelay
 		}
 
 		stx := &models.SignedTx{}