Enhanced validator details

.hack/devnet/run.sh

@@ -94,6 +94,7 @@ $(for node in $EXECUTION_NODES; do
 done)
 indexer:
   inMemoryEpochs: 8
+  activityHistoryLength: 6
   cachePersistenceDelay: 8
   disableIndexWriter: false
   syncEpochCooldown: 1

cmd/dora-explorer/main.go

@@ -175,6 +175,7 @@ func startFrontend(webserver *http.Server) {
 	if utils.Config.Frontend.Pprof {
 		// add pprof handler
 		router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux)
+		router.HandleFunc("/debug/cache", handlers.DebugCache).Methods("GET")
 	}
 
 	if utils.Config.Frontend.Debug {

config/default.config.yml

@@ -65,6 +65,9 @@ indexer:
   # max number of epochs to keep in memory
   inMemoryEpochs: 3
 
+  # number of epochs to keep validator activity history for (high memory usage for large validator sets)
+  activityHistoryLength: 6
+
   # disable synchronizing historic data
   disableSynchronizer: false
 

db/consolidation_request_txs.go

@@ -140,6 +140,11 @@ func GetConsolidationRequestTxsFiltered(offset uint64, limit uint32, canonicalFo
 		fmt.Fprintf(&sql, " %v dequeue_block <= $%v", filterOp, len(args))
 		filterOp = "AND"
 	}
+	if len(filter.PublicKey) > 0 {
+		args = append(args, filter.PublicKey)
+		fmt.Fprintf(&sql, " %v (source_pubkey = $%v OR target_pubkey = $%v) ", filterOp, len(args), len(args))
+		filterOp = "AND"
+	}
 	if len(filter.SourceAddress) > 0 {
 		args = append(args, filter.SourceAddress)
 		fmt.Fprintf(&sql, " %v source_address = $%v", filterOp, len(args))

db/consolidation_requests.go

@@ -60,7 +60,7 @@ func InsertConsolidationRequests(consolidations []*dbtypes.ConsolidationRequest,
 	return nil
 }
 
-func GetConsolidationRequestsFiltered(offset uint64, limit uint32, finalizedBlock uint64, filter *dbtypes.ConsolidationRequestFilter) ([]*dbtypes.ConsolidationRequest, uint64, error) {
+func GetConsolidationRequestsFiltered(offset uint64, limit uint32, canonicalForkIds []uint64, filter *dbtypes.ConsolidationRequestFilter) ([]*dbtypes.ConsolidationRequest, uint64, error) {
 	var sql strings.Builder
 	args := []interface{}{}
 	fmt.Fprint(&sql, `
@@ -92,6 +92,11 @@ func GetConsolidationRequestsFiltered(offset uint64, limit uint32, finalizedBloc
 		fmt.Fprintf(&sql, " %v slot_number <= $%v", filterOp, len(args))
 		filterOp = "AND"
 	}
+	if len(filter.PublicKey) > 0 {
+		args = append(args, filter.PublicKey)
+		fmt.Fprintf(&sql, " %v (source_pubkey = $%v OR target_pubkey = $%v) ", filterOp, len(args), len(args))
+		filterOp = "AND"
+	}
 	if len(filter.SourceAddress) > 0 {
 		args = append(args, filter.SourceAddress)
 		fmt.Fprintf(&sql, " %v source_address = $%v", filterOp, len(args))
@@ -136,14 +141,22 @@ func GetConsolidationRequestsFiltered(offset uint64, limit uint32, finalizedBloc
 		filterOp = "AND"
 	}
 
-	if filter.WithOrphaned == 0 {
-		args = append(args, finalizedBlock)
-		fmt.Fprintf(&sql, " %v (slot_number > $%v OR orphaned = false)", filterOp, len(args))
-		filterOp = "AND"
-	} else if filter.WithOrphaned == 2 {
-		args = append(args, finalizedBlock)
-		fmt.Fprintf(&sql, " %v (slot_number > $%v OR orphaned = true)", filterOp, len(args))
-		filterOp = "AND"
+	if filter.WithOrphaned != 1 {
+		forkIdStr := make([]string, len(canonicalForkIds))
+		for i, forkId := range canonicalForkIds {
+			forkIdStr[i] = fmt.Sprintf("%v", forkId)
+		}
+		if len(forkIdStr) == 0 {
+			forkIdStr = append(forkIdStr, "0")
+		}
+
+		if filter.WithOrphaned == 0 {
+			fmt.Fprintf(&sql, " %v fork_id IN (%v)", filterOp, strings.Join(forkIdStr, ","))
+			filterOp = "AND"
+		} else if filter.WithOrphaned == 2 {
+			fmt.Fprintf(&sql, " %v fork_id NOT IN (%v)", filterOp, strings.Join(forkIdStr, ","))
+			filterOp = "AND"
+		}
 	}
 
 	args = append(args, limit)

db/deposits.go

@@ -154,6 +154,16 @@ func GetDepositTxsFiltered(offset uint64, limit uint32, finalizedBlock uint64, f
 	`)
 
 	filterOp := "WHERE"
+	if filter.MinIndex > 0 {
+		args = append(args, filter.MinIndex)
+		fmt.Fprintf(&sql, " %v deposit_index >= $%v", filterOp, len(args))
+		filterOp = "AND"
+	}
+	if filter.MaxIndex > 0 {
+		args = append(args, filter.MaxIndex)
+		fmt.Fprintf(&sql, " %v deposit_index <= $%v", filterOp, len(args))
+		filterOp = "AND"
+	}
 	if len(filter.Address) > 0 {
 		args = append(args, filter.Address)
 		fmt.Fprintf(&sql, " %v tx_sender = $%v", filterOp, len(args))

db/withdrawal_request_txs.go

@@ -134,6 +134,11 @@ func GetWithdrawalRequestTxsFiltered(offset uint64, limit uint32, canonicalForkI
 		fmt.Fprintf(&sql, " %v dequeue_block <= $%v", filterOp, len(args))
 		filterOp = "AND"
 	}
+	if len(filter.PublicKey) > 0 {
+		args = append(args, filter.PublicKey)
+		fmt.Fprintf(&sql, " %v validator_pubkey = $%v ", filterOp, len(args))
+		filterOp = "AND"
+	}
 	if len(filter.SourceAddress) > 0 {
 		args = append(args, filter.SourceAddress)
 		fmt.Fprintf(&sql, " %v source_address = $%v", filterOp, len(args))

db/withdrawal_requests.go

@@ -61,7 +61,7 @@ func InsertWithdrawalRequests(elRequests []*dbtypes.WithdrawalRequest, tx *sqlx.
 	return nil
 }
 
-func GetWithdrawalRequestsFiltered(offset uint64, limit uint32, finalizedBlock uint64, filter *dbtypes.WithdrawalRequestFilter) ([]*dbtypes.WithdrawalRequest, uint64, error) {
+func GetWithdrawalRequestsFiltered(offset uint64, limit uint32, canonicalForkIds []uint64, filter *dbtypes.WithdrawalRequestFilter) ([]*dbtypes.WithdrawalRequest, uint64, error) {
 	var sql strings.Builder
 	args := []interface{}{}
 	fmt.Fprint(&sql, `
@@ -88,6 +88,11 @@ func GetWithdrawalRequestsFiltered(offset uint64, limit uint32, finalizedBlock u
 		fmt.Fprintf(&sql, " %v slot_number <= $%v", filterOp, len(args))
 		filterOp = "AND"
 	}
+	if len(filter.PublicKey) > 0 {
+		args = append(args, filter.PublicKey)
+		fmt.Fprintf(&sql, " %v validator_pubkey = $%v ", filterOp, len(args))
+		filterOp = "AND"
+	}
 	if len(filter.SourceAddress) > 0 {
 		args = append(args, filter.SourceAddress)
 		fmt.Fprintf(&sql, " %v source_address = $%v", filterOp, len(args))
@@ -123,14 +128,22 @@ func GetWithdrawalRequestsFiltered(offset uint64, limit uint32, finalizedBlock u
 		filterOp = "AND"
 	}
 
-	if filter.WithOrphaned == 0 {
-		args = append(args, finalizedBlock)
-		fmt.Fprintf(&sql, " %v (slot_number > $%v OR orphaned = false)", filterOp, len(args))
-		filterOp = "AND"
-	} else if filter.WithOrphaned == 2 {
-		args = append(args, finalizedBlock)
-		fmt.Fprintf(&sql, " %v (slot_number > $%v OR orphaned = true)", filterOp, len(args))
-		filterOp = "AND"
+	if filter.WithOrphaned != 1 {
+		forkIdStr := make([]string, len(canonicalForkIds))
+		for i, forkId := range canonicalForkIds {
+			forkIdStr[i] = fmt.Sprintf("%v", forkId)
+		}
+		if len(forkIdStr) == 0 {
+			forkIdStr = append(forkIdStr, "0")
+		}
+
+		if filter.WithOrphaned == 0 {
+			fmt.Fprintf(&sql, " %v fork_id IN (%v)", filterOp, strings.Join(forkIdStr, ","))
+			filterOp = "AND"
+		} else if filter.WithOrphaned == 2 {
+			fmt.Fprintf(&sql, " %v fork_id NOT IN (%v)", filterOp, strings.Join(forkIdStr, ","))
+			filterOp = "AND"
+		}
 	}
 
 	args = append(args, limit)

dbtypes/other.go

@@ -52,6 +52,8 @@ type MevBlockFilter struct {
 }
 
 type DepositTxFilter struct {
+	MinIndex      uint64
+	MaxIndex      uint64
 	Address       []byte
 	TargetAddress []byte
 	PublicKey     []byte
@@ -96,6 +98,7 @@ type SlashingFilter struct {
 type WithdrawalRequestFilter struct {
 	MinSlot       uint64
 	MaxSlot       uint64
+	PublicKey     []byte
 	SourceAddress []byte
 	MinIndex      uint64
 	MaxIndex      uint64
@@ -108,6 +111,7 @@ type WithdrawalRequestFilter struct {
 type WithdrawalRequestTxFilter struct {
 	MinDequeue    uint64
 	MaxDequeue    uint64
+	PublicKey     []byte
 	SourceAddress []byte
 	MinIndex      uint64
 	MaxIndex      uint64
@@ -120,6 +124,7 @@ type WithdrawalRequestTxFilter struct {
 type ConsolidationRequestFilter struct {
 	MinSlot          uint64
 	MaxSlot          uint64
+	PublicKey        []byte
 	SourceAddress    []byte
 	MinSrcIndex      uint64
 	MaxSrcIndex      uint64
@@ -133,6 +138,7 @@ type ConsolidationRequestFilter struct {
 type ConsolidationRequestTxFilter struct {
 	MinDequeue       uint64
 	MaxDequeue       uint64
+	PublicKey        []byte
 	SourceAddress    []byte
 	MinSrcIndex      uint64
 	MaxSrcIndex      uint64

go.mod

@@ -41,6 +41,7 @@ require (
 )
 
 require (
+	github.com/520MianXiangDuiXiang520/MapSize v0.0.0-20230414174449-030467540731 // indirect
 	github.com/ipfs/go-cid v0.4.1 // indirect
 	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect

go.sum

@@ -2,6 +2,8 @@ dario.cat/mergo v1.0.1 h1:Ra4+bf83h2ztPIQYNP99R6m+Y7KfnARDfID+a+vLl4s=
 dario.cat/mergo v1.0.1/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
+github.com/520MianXiangDuiXiang520/MapSize v0.0.0-20230414174449-030467540731 h1:ynhEKD+8g0G5Rk98HVqGeenoRBoHz0In6bAQ7bEzklQ=
+github.com/520MianXiangDuiXiang520/MapSize v0.0.0-20230414174449-030467540731/go.mod h1:YGrfzOdPf4xnEhFGmxjVSw0xuu/PLmGsJZipIpp8+C0=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/DataDog/zstd v1.5.5 h1:oWf5W7GtOLgp6bciQYDmhHHjdhYkALu6S/5Ni9ZgSvQ=
 github.com/DataDog/zstd v1.5.5/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=

handlers/debug_cache.go

@@ -0,0 +1,43 @@
+package handlers
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/ethpandaops/dora/services"
+	"github.com/ethpandaops/dora/templates"
+	"github.com/ethpandaops/dora/utils"
+)
+
+// DebugCache will submit a consolidation request
+func DebugCache(w http.ResponseWriter, r *http.Request) {
+	var debugCacheTemplateFiles = append(layoutTemplateFiles,
+		"debug_cache/debug_cache.html",
+	)
+	var pageTemplate = templates.GetTemplate(debugCacheTemplateFiles...)
+
+	if !utils.Config.Frontend.Pprof {
+		handlePageError(w, r, errors.New("debug pages are not enabled"))
+		return
+	}
+
+	pageData := buildDebugCachePageData()
+	data := InitPageData(w, r, "blockchain", "/debug_cache", "Debug Cache", debugCacheTemplateFiles)
+	data.Data = pageData
+	w.Header().Set("Content-Type", "text/html")
+	if handleTemplateError(w, r, "debug_cache.go", "Debug Cache", "", pageTemplate.ExecuteTemplate(w, "layout", data)) != nil {
+		return // an error has occurred and was processed
+	}
+}
+
+func buildDebugCachePageData() string {
+	logrus.Debugf("debug cache page called")
+
+	cacheStats := services.GlobalBeaconService.GetBeaconIndexer().GetCacheDebugStats()
+	jsonStats, _ := json.MarshalIndent(cacheStats, "", "  ")
+
+	return string(jsonStats)
+}

handlers/deposits.go

@@ -78,8 +78,6 @@ func buildDepositsPageData(firstEpoch uint64, pageSize uint64) (*models.Deposits
 	}
 
 	chainState := services.GlobalBeaconService.GetChainState()
-	validatorSetRsp := services.GlobalBeaconService.GetCachedValidatorPubkeyMap()
-	validatorActivityMap, validatorActivityMax := services.GlobalBeaconService.GetValidatorActivity(3, false)
 
 	// load initiated deposits
 	dbDepositTxs := db.GetDepositTxs(0, 20)
@@ -97,14 +95,14 @@ func buildDepositsPageData(firstEpoch uint64, pageSize uint64) (*models.Deposits
 			Valid:                 depositTx.ValidSignature,
 		}
 
-		validator := validatorSetRsp[phase0.BLSPubKey(depositTx.PublicKey)]
-		if validator == nil {
+		validatorIndex, found := services.GlobalBeaconService.GetValidatorIndexByPubkey(phase0.BLSPubKey(depositTx.PublicKey))
+		if !found {
 			depositTxData.ValidatorStatus = "Deposited"
 		} else {
+			validator := services.GlobalBeaconService.GetValidatorByIndex(validatorIndex, false)
 			if strings.HasPrefix(validator.Status.String(), "pending") {
 				depositTxData.ValidatorStatus = "Pending"
 			} else if validator.Status == v1.ValidatorStateActiveOngoing {
-				depositTxData.ValidatorStatus = "Active"
 				depositTxData.ShowUpcheck = true
 			} else if validator.Status == v1.ValidatorStateActiveExiting {
 				depositTxData.ValidatorStatus = "Exiting"
@@ -121,8 +119,8 @@ func buildDepositsPageData(firstEpoch uint64, pageSize uint64) (*models.Deposits
 			}
 
 			if depositTxData.ShowUpcheck {
-				depositTxData.UpcheckActivity = validatorActivityMap[validator.Index]
-				depositTxData.UpcheckMaximum = uint8(validatorActivityMax)
+				depositTxData.UpcheckActivity = uint8(services.GlobalBeaconService.GetValidatorLiveness(validator.Index, 3))
+				depositTxData.UpcheckMaximum = uint8(3)
 			}
 		}
 
@@ -148,10 +146,11 @@ func buildDepositsPageData(firstEpoch uint64, pageSize uint64) (*models.Deposits
 			depositData.Index = *deposit.Index
 		}
 
-		validator := validatorSetRsp[phase0.BLSPubKey(deposit.PublicKey)]
-		if validator == nil {
+		validatorIndex, found := services.GlobalBeaconService.GetValidatorIndexByPubkey(phase0.BLSPubKey(deposit.PublicKey))
+		if !found {
 			depositData.ValidatorStatus = "Deposited"
 		} else {
+			validator := services.GlobalBeaconService.GetValidatorByIndex(validatorIndex, false)
 			if strings.HasPrefix(validator.Status.String(), "pending") {
 				depositData.ValidatorStatus = "Pending"
 			} else if validator.Status == v1.ValidatorStateActiveOngoing {
@@ -172,8 +171,8 @@ func buildDepositsPageData(firstEpoch uint64, pageSize uint64) (*models.Deposits
 			}
 
 			if depositData.ShowUpcheck {
-				depositData.UpcheckActivity = validatorActivityMap[validator.Index]
-				depositData.UpcheckMaximum = uint8(validatorActivityMax)
+				depositData.UpcheckActivity = uint8(services.GlobalBeaconService.GetValidatorLiveness(validator.Index, 3))
+				depositData.UpcheckMaximum = uint8(3)
 			}
 		}