Merge branch 'staging' into BEDS-151/Add_RP_mode

.github/workflows/backend-linter.yml

@@ -37,7 +37,7 @@ jobs:
         with:
           version: v1.60.1
           working-directory: backend
-          args: --timeout=5m
+          args: --timeout=5m --config .golangci.yml
 
 
 

backend/.golangci.yml

@@ -20,6 +20,7 @@ linters:
     - importas
     - unparam
   disable:
+    - testpackage
     - dupword
     - gocognit
     - dupl

backend/cmd/evm_node_indexer/main.go

@@ -19,22 +19,22 @@ import (
 	"sync/atomic"
 	"time"
 
+	gcp_bigtable "cloud.google.com/go/bigtable"
 	"github.com/ethereum/go-ethereum"
 	gethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/gtuk/discordwebhook"
+	"github.com/lib/pq"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/api/option"
+
 	"github.com/gobitfly/beaconchain/pkg/commons/db"
 	"github.com/gobitfly/beaconchain/pkg/commons/hexutil"
 	"github.com/gobitfly/beaconchain/pkg/commons/log"
 	"github.com/gobitfly/beaconchain/pkg/commons/metrics"
 	"github.com/gobitfly/beaconchain/pkg/commons/types"
 	"github.com/gobitfly/beaconchain/pkg/commons/utils"
 	"github.com/gobitfly/beaconchain/pkg/commons/version"
-	"github.com/gtuk/discordwebhook"
-	"github.com/lib/pq"
-	"golang.org/x/sync/errgroup"
-	"google.golang.org/api/option"
-
-	gcp_bigtable "cloud.google.com/go/bigtable"
 )
 
 // defines
@@ -86,12 +86,18 @@ type intRange struct {
 
 // local globals
 var currentNodeBlockNumber atomic.Int64
-var elClient *ethclient.Client
+var elClient ethClient
 var reorgDepth *int64
 var httpClient *http.Client
 var errorIdentifier *regexp.Regexp
 var eth1RpcEndpoint string
 
+type ethClient interface {
+	SubscribeNewHead(ctx context.Context, ch chan<- *gethtypes.Header) (ethereum.Subscription, error)
+	ethereum.ChainIDReader
+	ethereum.BlockNumberReader
+}
+
 // init
 func init() {
 	httpClient = &http.Client{Timeout: time.Second * HTTP_TIMEOUT_IN_SECONDS}
@@ -275,7 +281,9 @@ func Run() {
 
 	// get latest block (as it's global, so we have a initial value)
 	log.Info("get latest block from node...")
-	updateBlockNumber(true, *noNewBlocks, time.Duration(*noNewBlocksThresholdSeconds)*time.Second, discordWebhookReportUrl, discordWebhookUser, discordWebhookAddTextFatal)
+	if err := updateBlockNumber(*noNewBlocks, time.Duration(*noNewBlocksThresholdSeconds)*time.Second, discordWebhookReportUrl, discordWebhookUser, discordWebhookAddTextFatal); err != nil {
+		log.Fatal(err, "updateBlockNumber", 0)
+	}
 	log.Infof("...get latest block (%s) from node done.", _formatInt64(currentNodeBlockNumber.Load()))
 
 	// //////////////////////////////////////////
@@ -892,21 +900,30 @@ func _splitAndVerifyJsonArray(jArray []byte, providedElementCount int64) ([][]by
 	return r, nil
 }
 
-// get newest block number from node, should be called always with TRUE
-func updateBlockNumber(firstCall bool, noNewBlocks bool, noNewBlocksThresholdDuration time.Duration, discordWebhookReportUrl *string, discordWebhookUser *string, discordWebhookAddTextFatal *string) {
-	if firstCall {
-		blockNumber, err := rpciGetLatestBlock()
-		if err != nil {
-			sendMessage(fmt.Sprintf("%s NODE EXPORT: Fatal, failed to get newest block from node, on first try %s", getChainNamePretty(), *discordWebhookAddTextFatal), discordWebhookReportUrl, discordWebhookUser)
-			log.Fatal(err, "fatal, failed to get newest block from node, on first try", 0)
-		}
-		currentNodeBlockNumber.Store(blockNumber)
-		if !noNewBlocks {
-			go updateBlockNumber(false, false, noNewBlocksThresholdDuration, discordWebhookReportUrl, discordWebhookUser, discordWebhookAddTextFatal)
-		}
-		return
+// get newest block number from node
+func updateBlockNumber(noNewBlocks bool, noNewBlocksThresholdDuration time.Duration, discordWebhookReportUrl *string, discordWebhookUser *string, discordWebhookAddTextFatal *string) error {
+	blockNumber, err := rpciGetLatestBlock()
+	if err != nil {
+		sendMessage(fmt.Sprintf("%s NODE EXPORT: Fatal, failed to get newest block from node, on first try %s", getChainNamePretty(), *discordWebhookAddTextFatal), discordWebhookReportUrl, discordWebhookUser)
+		log.Fatal(err, "fatal, failed to get newest block from node, on first try", 0)
 	}
+	currentNodeBlockNumber.Store(blockNumber)
+	// request the block number a second time to verify if node is not rewinding
+	blockNumber, err = rpciGetLatestBlock()
+	if err != nil {
+		return err
+	}
+	if blockNumber < currentNodeBlockNumber.Load() {
+		return fmt.Errorf("node is rewinding: newest block %d < previous block %d", blockNumber, currentNodeBlockNumber.Load())
+	}
+
+	if !noNewBlocks {
+		go monitorBlockNumber(noNewBlocksThresholdDuration, discordWebhookReportUrl, discordWebhookUser, discordWebhookAddTextFatal)
+	}
+	return nil
+}
 
+func monitorBlockNumber(noNewBlocksThresholdDuration time.Duration, discordWebhookReportUrl *string, discordWebhookUser *string, discordWebhookAddTextFatal *string) {
 	var errorText string
 	gotNewBlockAt := time.Now()
 	timePerBlock := time.Second * time.Duration(utils.Config.Chain.ClConfig.SecondsPerSlot)

backend/cmd/evm_node_indexer/main_test.go

@@ -0,0 +1,55 @@
+package evm_node_indexer
+
+import (
+	"context"
+	"math/big"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum"
+	gethtypes "github.com/ethereum/go-ethereum/core/types"
+
+	"github.com/gobitfly/beaconchain/pkg/commons/types"
+	"github.com/gobitfly/beaconchain/pkg/commons/utils"
+)
+
+func TestUpdateBlockNumberWithRewindingNode(t *testing.T) {
+	utils.Config = &types.Config{
+		Chain: types.Chain{
+			ClConfig: types.ClChainConfig{
+				SecondsPerSlot: 0,
+			},
+		},
+	}
+	elClient = &fakeEthClient{
+		chainID: big.NewInt(1),
+		blocks:  []uint64{126192682, 126186030},
+	}
+	err := updateBlockNumber(false, time.Second, nil, nil, nil)
+	if err == nil {
+		t.Fatal("expected error got nil")
+	}
+	if !strings.Contains(err.Error(), "node is rewinding") {
+		t.Fatalf("error message does not contain 'node is rewinding', got %s", err.Error())
+	}
+}
+
+type fakeEthClient struct {
+	chainID *big.Int
+	blocks  []uint64
+}
+
+func (f *fakeEthClient) SubscribeNewHead(ctx context.Context, ch chan<- *gethtypes.Header) (ethereum.Subscription, error) {
+	panic("implement me")
+}
+
+func (f *fakeEthClient) ChainID(ctx context.Context) (*big.Int, error) {
+	return f.chainID, nil
+}
+
+func (f *fakeEthClient) BlockNumber(ctx context.Context) (uint64, error) {
+	block := f.blocks[0]
+	f.blocks = f.blocks[1:]
+	return block, nil
+}

backend/cmd/misc/main.go

@@ -565,19 +565,19 @@ func collectNotifications(startEpoch uint64) error {
 		spew.Dump(notifications[0])
 	}
 
-	emails, err := notification.RenderEmailsForUserEvents(0, notifications)
-	if err != nil {
-		return err
-	}
+	// emails, err := notification.RenderEmailsForUserEvents(0, notifications)
+	// if err != nil {
+	// 	return err
+	// }
 
-	for _, email := range emails {
-		// if email.Address == "" {
-		log.Infof("to: %v", email.Address)
-		log.Infof("subject: %v", email.Subject)
-		log.Infof("body: %v", email.Email.Body)
-		log.Info("-----")
-		// }
-	}
+	// for _, email := range emails {
+	// 	// if email.Address == "" {
+	// 	log.Infof("to: %v", email.Address)
+	// 	log.Infof("subject: %v", email.Subject)
+	// 	log.Infof("body: %v", email.Email.Body)
+	// 	log.Info("-----")
+	// 	// }
+	// }
 
 	// pushMessages, err := notification.RenderPushMessagesForUserEvents(0, notifications)
 	// if err != nil {

backend/go.mod

@@ -193,6 +193,7 @@ require (
 	github.com/jbenet/goprocess v0.1.4 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/k3a/html2text v1.2.1 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
 	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect

backend/go.sum

@@ -597,6 +597,8 @@ github.com/juliangruber/go-intersect v1.1.0/go.mod h1:WMau+1kAmnlQnKiikekNJbtGtf
 github.com/jung-kurt/gofpdf v1.0.0/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/jung-kurt/gofpdf v1.16.2 h1:jgbatWHfRlPYiK85qgevsZTHviWXKwB1TTiKdz5PtRc=
 github.com/jung-kurt/gofpdf v1.16.2/go.mod h1:1hl7y57EsiPAkLbOwzpzqgx1A30nQCk/YmFV8S2vmK0=
+github.com/k3a/html2text v1.2.1 h1:nvnKgBvBR/myqrwfLuiqecUtaK1lB9hGziIJKatNFVY=
+github.com/k3a/html2text v1.2.1/go.mod h1:ieEXykM67iT8lTvEWBh6fhpH4B23kB9OMKPdIBmgUqA=
 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs=
 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
 github.com/kelseyhightower/envconfig v1.4.0 h1:Im6hONhd3pLkfDFsbRgu68RDNkGF1r3dvMUtDTo2cv8=

backend/pkg/api/data_access/archiver.go

@@ -91,7 +91,7 @@ func (d *DataAccessService) UpdateValidatorDashboardsArchiving(ctx context.Conte
 
 	query, args, err := ds.Prepared(true).ToSQL()
 	if err != nil {
-		return fmt.Errorf("error preparing query: %v", err)
+		return fmt.Errorf("error preparing query: %w", err)
 	}
 
 	_, err = d.writerDb.ExecContext(ctx, query, args...)

backend/pkg/api/data_access/dummy.go

@@ -258,6 +258,10 @@ func (d *DummyService) RemoveValidatorDashboardGroup(ctx context.Context, dashbo
 	return nil
 }
 
+func (d *DummyService) RemoveValidatorDashboardGroupValidators(ctx context.Context, dashboardId t.VDBIdPrimary, groupId uint64) error {
+	return nil
+}
+
 func (d *DummyService) GetValidatorDashboardGroupExists(ctx context.Context, dashboardId t.VDBIdPrimary, groupId uint64) (bool, error) {
 	return true, nil
 }
@@ -563,9 +567,6 @@ func (d *DummyService) GetMachineNotifications(ctx context.Context, userId uint6
 func (d *DummyService) GetClientNotifications(ctx context.Context, userId uint64, cursor string, colSort t.Sort[enums.NotificationClientsColumn], search string, limit uint64) ([]t.NotificationClientsTableRow, *t.Paging, error) {
 	return getDummyWithPaging[t.NotificationClientsTableRow](ctx)
 }
-func (d *DummyService) GetRocketPoolNotifications(ctx context.Context, userId uint64, cursor string, colSort t.Sort[enums.NotificationRocketPoolColumn], search string, limit uint64) ([]t.NotificationRocketPoolTableRow, *t.Paging, error) {
-	return getDummyWithPaging[t.NotificationRocketPoolTableRow](ctx)
-}
 func (d *DummyService) GetNetworkNotifications(ctx context.Context, userId uint64, cursor string, colSort t.Sort[enums.NotificationNetworksColumn], limit uint64) ([]t.NotificationNetworksTableRow, *t.Paging, error) {
 	return getDummyWithPaging[t.NotificationNetworksTableRow](ctx)
 }

backend/pkg/api/data_access/general.go

@@ -2,6 +2,7 @@ package dataaccess
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/doug-martin/goqu/v9"
 	"github.com/doug-martin/goqu/v9/exp"
@@ -52,7 +53,7 @@ func (d *DataAccessService) GetNamesAndEnsForAddresses(ctx context.Context, addr
 // helper function to sort and apply pagination to a query
 // 1st param is the list of all columns necessary to sort the table deterministically; it defines their precedence and sort direction
 // 2nd param is the requested sort column; it may or may not be part of the default columns (if it is, you don't have to specify the cursor limit again)
-func applySortAndPagination(defaultColumns []types.SortColumn, primary types.SortColumn, cursor types.GenericCursor) ([]exp.OrderedExpression, exp.Expression) {
+func applySortAndPagination(defaultColumns []types.SortColumn, primary types.SortColumn, cursor types.GenericCursor) ([]exp.OrderedExpression, exp.Expression, error) {
 	// prepare ordering columns; always need all columns to ensure consistent ordering
 	queryOrderColumns := make([]types.SortColumn, 0, len(defaultColumns))
 	queryOrderColumns = append(queryOrderColumns, primary)
@@ -90,22 +91,43 @@ func applySortAndPagination(defaultColumns []types.SortColumn, primary types.Sor
 			var colWhere exp.Expression
 
 			// current convention is opposite of the psql default (ASC: nulls first, DESC: nulls last)
-			colWhere = goqu.Or(column.Column.Lt(column.Offset), column.Column.IsNull())
-			if !column.Desc {
-				colWhere = column.Column.Gt(column.Offset)
-				if column.Offset == nil {
-					colWhere = goqu.Or(colWhere, column.Column.IsNull())
+			if column.Desc {
+				if column.Offset == nil && queryWhere == nil {
+					continue
 				}
-			}
 
-			if queryWhere == nil {
-				queryWhere = colWhere
+				colWhere = goqu.Or(column.Column.Lt(column.Offset), column.Column.IsNull())
+
+				if queryWhere == nil {
+					queryWhere = colWhere
+				} else {
+					if column.Offset == nil {
+						queryWhere = goqu.And(column.Column.IsNull(), queryWhere)
+					} else {
+						queryWhere = goqu.And(column.Column.Eq(column.Offset), queryWhere)
+						queryWhere = goqu.Or(colWhere, queryWhere)
+					}
+				}
 			} else {
-				queryWhere = goqu.And(column.Column.Eq(column.Offset), queryWhere)
-				queryWhere = goqu.Or(colWhere, queryWhere)
+				if column.Offset == nil {
+					colWhere = column.Column.IsNotNull()
+				} else {
+					colWhere = column.Column.Gt(column.Offset)
+				}
+
+				if queryWhere == nil {
+					queryWhere = colWhere
+				} else {
+					queryWhere = goqu.And(column.Column.Eq(column.Offset), queryWhere)
+					queryWhere = goqu.Or(colWhere, queryWhere)
+				}
 			}
 		}
+
+		if queryWhere == nil {
+			return nil, nil, fmt.Errorf("cursor given for descending order but all offset are nil meaning no data after it")
+		}
 	}
 
-	return queryOrder, queryWhere
+	return queryOrder, queryWhere, nil
 }