feat(eventsyncing): health calls check last seen block distance

codecov.yml

@@ -39,6 +39,7 @@ ignore:
   - "network/p2p/metrics.go"
   - "network/peers/metrics.go"
   - "network/peers/connections/metrics.go"
+  - "network/peers/connections/mock/mock_storage.go"
   - "network/topics/metrics.go"
   - "protocol/v2/ssv/validator/metrics.go"
   - "protocol/v2/ssv/runner/metrics/metrics.go"
@@ -47,6 +48,7 @@ ignore:
   - "exporter/api/metrics.go"
   - "eth/executionclient/metrics.go"
   - "eth/eventsyncer/metrics.go"
+  - "eth/eventsyncer/event_syncer_mock.go"
   - "eth/eventhandler/metrics.go"
   - "operator/validator/metrics.go"
   - "operator/metrics.go"

eth/eventsyncer/event_syncer.go

@@ -6,15 +6,19 @@
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
 	"time"
 
+	"github.com/ethereum/go-ethereum/core/types"
 	"go.uber.org/zap"
 
 	"github.com/ssvlabs/ssv/eth/executionclient"
 	"github.com/ssvlabs/ssv/logging/fields"
 	nodestorage "github.com/ssvlabs/ssv/operator/storage"
 )
 
+//go:generate mockgen -package=eventsyncer -destination=./event_syncer_mock.go -source=./event_syncer.go
+
 // TODO: check if something from these PRs need to be ported:
 // https://github.com/ssvlabs/ssv/pull/1053
 
@@ -26,6 +30,7 @@
 type ExecutionClient interface {
 	FetchHistoricalLogs(ctx context.Context, fromBlock uint64) (logs <-chan executionclient.BlockLogs, errors <-chan error, err error)
 	StreamLogs(ctx context.Context, fromBlock uint64) <-chan executionclient.BlockLogs
+	HeaderByNumber(ctx context.Context, blockNumber *big.Int) (*types.Header, error)
 }
 
 type EventHandler interface {
@@ -80,6 +85,21 @@
 	if time.Since(es.lastProcessedBlockChange) > es.stalenessThreshold {
 		return fmt.Errorf("syncing is stuck at block %d", lastProcessedBlock.Uint64())
 	}
+
+	return es.blockBelowThreshold(ctx, lastProcessedBlock)
+}
+
+func (es *EventSyncer) blockBelowThreshold(ctx context.Context, block *big.Int) error {
+	header, err := es.executionClient.HeaderByNumber(ctx, block)
+	if err != nil {
+		return fmt.Errorf("failed to get header for block %d: %w", block, err)
+	}
+
+	// #nosec G115
+	if header.Time > 0 && header.Time < uint64(time.Now().Add(-es.stalenessThreshold).Unix()) {
+		return fmt.Errorf("block %d is too old", block)
+	}
+
 	return nil
 }
 
@@ -110,6 +130,12 @@
 		return 0, fmt.Errorf("event replay: lastProcessedBlock (%d) is lower than fromBlock (%d)", lastProcessedBlock, fromBlock)
 	}
 
+	// Check if the block is too old.
+	b := new(big.Int).SetUint64(lastProcessedBlock)
+	if err := es.blockBelowThreshold(ctx, b); err != nil {
+		return 0, err
+	}
+
 	es.logger.Info("finished syncing historical events",
 		zap.Uint64("from_block", fromBlock),
 		zap.Uint64("last_processed_block", lastProcessedBlock))

eth/eventsyncer/event_syncer_test.go

@@ -3,6 +3,7 @@ package eventsyncer
 import (
 	"context"
 	"encoding/base64"
+	"errors"
 	"math/big"
 	"net/http/httptest"
 	"strings"
@@ -12,7 +13,6 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	ethcommon "github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/types"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethclient/simulated"
@@ -197,7 +197,7 @@ func setupEventHandler(
 
 func simTestBackend(testAddr ethcommon.Address) *simulator.Backend {
 	return simulator.NewBackend(
-		types.GenesisAlloc{
+		ethtypes.GenesisAlloc{
 			testAddr: {Balance: big.NewInt(10000000000000000)},
 		}, simulated.WithBlockGasLimit(10000000),
 	)
@@ -240,3 +240,32 @@ func setupOperatorStorage(logger *zap.Logger, db basedb.Database, privKey keys.O
 
 	return nodeStorage, operatorData
 }
+
+func TestBlockBelowThreshold(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	m := NewMockExecutionClient(ctrl)
+	ctx := context.Background()
+
+	s := New(nil, m, nil)
+
+	t.Run("fails on EC error", func(t *testing.T) {
+		err1 := errors.New("ec err")
+		m.EXPECT().HeaderByNumber(ctx, big.NewInt(1)).Return(nil, err1)
+		err := s.blockBelowThreshold(ctx, big.NewInt(1))
+		require.ErrorIs(t, err, err1)
+	})
+
+	t.Run("fails if outside threashold", func(t *testing.T) {
+		header := &ethtypes.Header{Time: uint64(time.Now().Add(-151 * time.Second).Unix())}
+		m.EXPECT().HeaderByNumber(ctx, big.NewInt(1)).Return(header, nil)
+		err := s.blockBelowThreshold(ctx, big.NewInt(1))
+		require.Error(t, err)
+	})
+
+	t.Run("success", func(t *testing.T) {
+		header := &ethtypes.Header{Time: uint64(time.Now().Add(-149 * time.Second).Unix())}
+		m.EXPECT().HeaderByNumber(ctx, big.NewInt(1)).Return(header, nil)
+		err := s.blockBelowThreshold(ctx, big.NewInt(1))
+		require.NoError(t, err)
+	})
+}

eth/executionclient/execution_client.go

@@ -178,14 +178,17 @@
 					}
 					validLogs = append(validLogs, log)
 				}
-				if len(validLogs) == 0 {
-					// Emit empty block logs to indicate that we have advanced to this block.
-					logs <- BlockLogs{BlockNumber: toBlock}
-				} else {
-					for _, blockLogs := range PackLogs(validLogs) {
-						logs <- blockLogs
+				var highestBlock uint64
+				for _, blockLogs := range PackLogs(validLogs) {
+					logs <- blockLogs
+					if blockLogs.BlockNumber > highestBlock {
+						highestBlock = blockLogs.BlockNumber
 					}
 				}
+				// Emit empty block logs to indicate that we have advanced to this block.
+				if highestBlock < toBlock {
+					logs <- BlockLogs{BlockNumber: toBlock}
+				}
 			}
 		}
 	}()
@@ -292,6 +295,18 @@
 	return b, nil
 }
 
+func (ec *ExecutionClient) HeaderByNumber(ctx context.Context, blockNumber *big.Int) (*ethtypes.Header, error) {
+	h, err := ec.client.HeaderByNumber(ctx, blockNumber)
+	if err != nil {
+		ec.logger.Error(elResponseErrMsg,
+			zap.String("method", "eth_getBlockByNumber"),
+			zap.Error(err))
+		return nil, err
+	}
+
+	return h, nil
+}
+
 func (ec *ExecutionClient) isClosed() bool {
 	select {
 	case <-ec.closed:

network/peers/connections/mock/mock_storage.go

@@ -51,6 +51,16 @@ func (m NodeStorage) GetLastProcessedBlock(txn basedb.Reader) (*big.Int, bool, e
 	panic("implement me")
 }
 
+func (m NodeStorage) SaveHighestSeenBlock(txn basedb.ReadWriter, offset *big.Int) error {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (m NodeStorage) GetHighestSeenBlock(txn basedb.Reader) (*big.Int, bool, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (m NodeStorage) DropRegistryData() error {
 	//TODO implement me
 	panic("implement me")

operator/storage/storage.go

@@ -21,6 +21,7 @@
 var (
 	storagePrefix         = []byte("operator/")
 	lastProcessedBlockKey = []byte("syncOffset") // TODO: temporarily left as syncOffset for compatibility, consider renaming and adding a migration for that
+	highestSeenBlockKey   = []byte("lastSeen")
 	configKey             = []byte("config")
 )
 
@@ -176,6 +177,10 @@
 	return s.db.Using(rw).Set(storagePrefix, lastProcessedBlockKey, offset.Bytes())
 }
 
+func (s *storage) SaveHighestSeenBlock(rw basedb.ReadWriter, offset *big.Int) error {
+	return s.db.Using(rw).Set(storagePrefix, highestSeenBlockKey, offset.Bytes())
+}
+
 func (s *storage) dropLastProcessedBlock() error {
 	return s.db.DropPrefix(append(storagePrefix, lastProcessedBlockKey...))
 }
@@ -206,6 +211,20 @@
 	return offset, found, nil
 }
 
+// GetHighestSeenBlock returns the highest received block with or without events.
+func (s *storage) GetHighestSeenBlock(r basedb.Reader) (*big.Int, bool, error) {
+	obj, found, err := s.db.UsingReader(r).Get(storagePrefix, highestSeenBlockKey)
+	if !found {
+		return nil, found, nil
+	}
+	if err != nil {
+		return nil, found, err
+	}
+
+	offset := new(big.Int).SetBytes(obj.Value)
+	return offset, found, nil
+}
+
 // GetPrivateKeyHash return sha256 hashed private key
 func (s *storage) GetPrivateKeyHash() (string, bool, error) {
 	obj, found, err := s.db.Get(storagePrefix, []byte(HashedPrivateKey))