feat(BEDS-536): add re-index-blocks cmd to fix internal transfers status

cache/tiered_cache.go

@@ -13,6 +13,8 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+var _ TieredCacher = (*tieredCache)(nil)
+
 // Tiered cache is a cache implementation combining a
 type tieredCache struct {
 	localGoCache *freecache.Cache
@@ -31,7 +33,19 @@ type RemoteCache interface {
 	GetBool(ctx context.Context, key string) (bool, error)
 }
 
-var TieredCache *tieredCache
+type TieredCacher interface {
+	Set(key string, value interface{}, expiration time.Duration) error
+	SetString(key string, value string, expiration time.Duration) error
+	SetUint64(key string, value uint64, expiration time.Duration) error
+	SetBool(key string, value bool, expiration time.Duration) error
+
+	GetStringWithLocalTimeout(key string, localExpiration time.Duration) (string, error)
+	GetUint64WithLocalTimeout(key string, localExpiration time.Duration) (uint64, error)
+	GetBoolWithLocalTimeout(key string, localExpiration time.Duration) (bool, error)
+	GetWithLocalTimeout(key string, localExpiration time.Duration, returnValue interface{}) (interface{}, error)
+}
+
+var TieredCache TieredCacher
 
 func MustInitTieredCache(redisAddress string) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)

db/bigtable.go

@@ -67,7 +67,7 @@ type Bigtable struct {
 
 	tableMachineMetrics *gcp_bigtable.Table
 
-	redisCache *redis.Client
+	redisCache RedisClient
 
 	LastAttestationCache    map[uint64]uint64
 	LastAttestationCacheMux *sync.Mutex
@@ -79,10 +79,30 @@ type Bigtable struct {
 	machineMetricsQueuedWritesChan chan (types.BulkMutation)
 }
 
+type RedisClient interface {
+	SCard(ctx context.Context, key string) *redis.IntCmd
+	SetNX(ctx context.Context, key string, value interface{}, expiration time.Duration) *redis.BoolCmd
+	Pipeline() redis.Pipeliner
+	Get(ctx context.Context, key string) *redis.StringCmd
+	Set(ctx context.Context, key string, value interface{}, expiration time.Duration) *redis.StatusCmd
+}
+
 func InitBigtable(project, instance, chainId, redisAddress string) (*Bigtable, error) {
+	rdc := redis.NewClient(&redis.Options{
+		Addr:        redisAddress,
+		ReadTimeout: time.Second * 20,
+	})
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
+	defer cancel()
+	if err := rdc.Ping(ctx).Err(); err != nil {
+		return nil, err
+	}
 
-	if utils.Config.Bigtable.Emulator {
+	return InitBigtableWithCache(ctx, project, instance, chainId, rdc)
+}
 
+func InitBigtableWithCache(ctx context.Context, project, instance, chainId string, rdc RedisClient) (*Bigtable, error) {
+	if utils.Config.Bigtable.Emulator {
 		if utils.Config.Bigtable.EmulatorHost == "" {
 			utils.Config.Bigtable.EmulatorHost = "127.0.0.1"
 		}
@@ -93,26 +113,14 @@ func InitBigtable(project, instance, chainId, redisAddress string) (*Bigtable, e
 			logger.Fatalf("unable to set bigtable emulator environment variable: %v", err)
 		}
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
-	defer cancel()
 
 	poolSize := 50
 	btClient, err := gcp_bigtable.NewClient(ctx, project, instance, option.WithGRPCConnectionPool(poolSize))
-	// btClient, err := gcp_bigtable.NewClient(context.Background(), project, instance)
 
 	if err != nil {
 		return nil, err
 	}
 
-	rdc := redis.NewClient(&redis.Options{
-		Addr:        redisAddress,
-		ReadTimeout: time.Second * 20,
-	})
-
-	if err := rdc.Ping(ctx).Err(); err != nil {
-		return nil, err
-	}
-
 	bt := &Bigtable{
 		client:                         btClient,
 		tableData:                      btClient.Open("data"),

db/bigtable_eth1.go

@@ -11,6 +11,7 @@ import (
 	"log"
 	"math/big"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -24,8 +25,6 @@ import (
 	"github.com/gobitfly/eth2-beaconchain-explorer/types"
 	"github.com/gobitfly/eth2-beaconchain-explorer/utils"
 
-	"strconv"
-
 	gcp_bigtable "cloud.google.com/go/bigtable"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -1015,6 +1014,7 @@ func (bigtable *Bigtable) TransformTx(blk *types.Eth1Block, cache *freecache.Cac
 			BlobGasPrice:       tx.GetBlobGasPrice(),
 			IsContractCreation: isContract,
 			ErrorMsg:           tx.GetErrorMsg(),
+			Status:             tx.Status,
 		}
 		// Mark Sender and Recipient for balance update
 		bigtable.markBalanceUpdate(indexedTx.From, []byte{0x0}, bulkMetadataUpdates, cache)
@@ -1305,6 +1305,7 @@ func (bigtable *Bigtable) TransformItx(blk *types.Eth1Block, cache *freecache.Ca
 				From:        itx.GetFrom(),
 				To:          itx.GetTo(),
 				Value:       itx.GetValue(),
+				Reverted:    itx.GetReverted(),
 			}
 
 			bigtable.markBalanceUpdate(indexedItx.To, []byte{0x0}, bulkMetadataUpdates, cache)
@@ -2346,7 +2347,7 @@ func (bigtable *Bigtable) GetAddressTransactionsTableData(address []byte, pageTo
 		}
 
 		tableData[i] = []interface{}{
-			utils.FormatTransactionHash(t.Hash, t.ErrorMsg == ""),
+			utils.FormatTransactionHashFromStatus(t.Hash, t.Status),
 			utils.FormatMethod(bigtable.GetMethodLabel(t.MethodId, contractInteraction)),
 			utils.FormatBlockNumber(t.BlockNumber),
 			utils.FormatTimestamp(t.Time.AsTime().Unix()),
@@ -2825,7 +2826,7 @@ func (bigtable *Bigtable) GetAddressInternalTableData(address []byte, pageToken
 		}
 
 		tableData[i] = []interface{}{
-			utils.FormatTransactionHash(t.ParentHash, true),
+			utils.FormatTransactionHash(t.ParentHash, !t.Reverted),
 			utils.FormatBlockNumber(t.BlockNumber),
 			utils.FormatTimestamp(t.Time.AsTime().Unix()),
 			utils.FormatAddressWithLimitsInAddressPageTable(address, t.From, BigtableClient.GetAddressLabel(fromName, from_contractInteraction), from_contractInteraction != types.CONTRACT_NONE, digitLimitInAddressPagesTable, nameLimitInAddressPagesTable, true),
@@ -2869,7 +2870,20 @@ func (bigtable *Bigtable) GetInternalTransfersForTransaction(transaction []byte,
 	}
 
 	data := make([]types.ITransaction, 0, len(parityTrace)-1)
+	var revertSource []int64
 	for i := 1; i < len(parityTrace); i++ {
+		var reverted bool
+		if parityTrace[i].Error != "" {
+			reverted = true
+			// only save the highest root revert
+			if !isSubset(parityTrace[i].TraceAddress, revertSource) {
+				revertSource = parityTrace[i].TraceAddress
+			}
+		}
+		if isSubset(parityTrace[i].TraceAddress, revertSource) {
+			reverted = true
+		}
+
 		from, to, value, tx_type := parityTrace[i].ConvertFields()
 		if tx_type == "suicide" {
 			// erigon's "suicide" might be misleading for users
@@ -2896,8 +2910,9 @@ func (bigtable *Bigtable) GetInternalTransfersForTransaction(transaction []byte,
 			From:      utils.FormatAddress(from, nil, fromName, false, from_contractInteraction != types.CONTRACT_NONE, true),
 			To:        utils.FormatAddress(to, nil, toName, false, to_contractInteraction != types.CONTRACT_NONE, true),
 			Amount:    utils.FormatElCurrency(value, currency, 8, true, false, false, true),
-			TracePath: utils.FormatTracePath(tx_type, parityTrace[i].TraceAddress, parityTrace[i].Error == "", bigtable.GetMethodLabel(input, from_contractInteraction)),
+			TracePath: utils.FormatTracePath(tx_type, parityTrace[i].TraceAddress, !reverted, bigtable.GetMethodLabel(input, from_contractInteraction)),
 			Advanced:  tx_type == "delegatecall" || string(value) == "\x00",
+			Reverted:  reverted,
 		}
 
 		gaslimit, err := strconv.ParseUint(parityTrace[i].Action.Gas, 0, 0)

db/bigtable_init.go

@@ -10,6 +10,8 @@ import (
 	gcp_bigtable "cloud.google.com/go/bigtable"
 )
 
+var ErrTableAlreadyExist = fmt.Errorf("aborting bigtable schema init as tables are already present")
+
 func InitBigtableSchema() error {
 
 	tables := make(map[string]map[string]gcp_bigtable.GCPolicy)
@@ -62,7 +64,7 @@ func InitBigtableSchema() error {
 	}
 
 	if len(existingTables) > 0 {
-		return fmt.Errorf("aborting bigtable schema init as tables are already present")
+		return ErrTableAlreadyExist
 	}
 
 	for name, definition := range tables {

db/db.go

@@ -2,6 +2,7 @@ package db
 
 import (
 	"bytes"
+	"context"
 	"database/sql"
 	"embed"
 	"encoding/hex"
@@ -37,9 +38,19 @@ var EmbedMigrations embed.FS
 
 var DBPGX *pgxpool.Conn
 
+type SQLReaderDb interface {
+	Close() error
+	Get(dest interface{}, query string, args ...interface{}) error
+	Select(dest interface{}, query string, args ...interface{}) error
+	SelectContext(ctx context.Context, dest interface{}, query string, args ...interface{}) error
+	Query(query string, args ...any) (*sql.Rows, error)
+	Preparex(query string) (*sqlx.Stmt, error)
+	Rebind(query string) string
+}
+
 // DB is a pointer to the explorer-database
 var WriterDb *sqlx.DB
-var ReaderDb *sqlx.DB
+var ReaderDb SQLReaderDb
 
 var logger = logrus.StandardLogger().WithField("module", "db")
 

db/statistics.go

@@ -1637,7 +1637,7 @@ func WriteExecutionChartSeriesForDay(day int64) error {
 				totalBlobCount = totalBlobCount.Add(decimal.NewFromInt(int64(len(tx.BlobVersionedHashes))))
 
 			default:
-				logger.Fatalf("error unknown tx type %v hash: %x", tx.Status, tx.Hash)
+				logger.Fatalf("error unknown tx type %v hash: %x", tx.Type, tx.Hash)
 			}
 			totalTxFees = totalTxFees.Add(txFees)
 
@@ -1646,7 +1646,7 @@ func WriteExecutionChartSeriesForDay(day int64) error {
 				failedTxCount += 1
 				totalFailedGasUsed = totalFailedGasUsed.Add(gasUsed)
 				totalFailedTxFee = totalFailedTxFee.Add(txFees)
-			case 1:
+			case 1, 2:
 				successTxCount += 1
 			default:
 				logger.Fatalf("error unknown status code %v hash: %x", tx.Status, tx.Hash)

db/utils.go

@@ -0,0 +1,16 @@
+package db
+
+func isSubset[E comparable](big []E, short []E) bool {
+	if len(short) == 0 {
+		return false
+	}
+	if len(big) < len(short) {
+		return false
+	}
+	for i := 0; i < len(short); i++ {
+		if big[i] != short[i] {
+			return false
+		}
+	}
+	return true
+}