Skip to content

Commit 6385287

Browse files
authored
Merge pull request #451 from ethpandaops/pk910/fork-visualization
chain fork visualization
2 parents 81e197f + f0459b9 commit 6385287

File tree

23 files changed

+4429
-14
lines changed

23 files changed

+4429
-14
lines changed

cmd/dora-explorer/main.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -179,6 +179,7 @@ func startFrontend(router *mux.Router) {
179179
router.HandleFunc("/clients/execution/refresh", handlers.ClientsELRefresh).Methods("POST")
180180
router.HandleFunc("/clients/execution/refresh/status", handlers.ClientsELRefreshStatus).Methods("GET")
181181
router.HandleFunc("/forks", handlers.Forks).Methods("GET")
182+
router.HandleFunc("/chain-forks", handlers.ChainForks).Methods("GET")
182183
router.HandleFunc("/epochs", handlers.Epochs).Methods("GET")
183184
router.HandleFunc("/epoch/{epoch}", handlers.Epoch).Methods("GET")
184185
router.HandleFunc("/slots", handlers.Slots).Methods("GET")

cmd/dora-utils/migrate.go

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"flag"
55
"fmt"
66
"os"
7+
"slices"
78
"strings"
89

910
"github.com/jmoiron/sqlx"
@@ -37,6 +38,8 @@ func migrate() {
3738
targetPgsqlPass := flags.String("target-pgsql-pass", "", "Target PostgreSQL password")
3839
targetPgsqlDb := flags.String("target-pgsql-db", "", "Target PostgreSQL database name")
3940

41+
limitTablesStr := flags.String("limit-tables", "", "Limit tables to migrate (comma separated list)")
42+
4043
debug := flags.Bool("debug", false, "Enable debug mode")
4144

4245
flags.Parse(os.Args[1:])
@@ -50,6 +53,11 @@ func migrate() {
5053
os.Exit(1)
5154
}
5255

56+
limitTables := make([]string, 0)
57+
if limitTablesStr != nil && *limitTablesStr != "" {
58+
limitTables = strings.Split(*limitTablesStr, ",")
59+
}
60+
5361
sourceConfig := DbConfig{Engine: *sourceEngine}
5462
targetConfig := DbConfig{Engine: *targetEngine}
5563

@@ -73,12 +81,12 @@ func migrate() {
7381
targetConfig.Pgsql.Name = *targetPgsqlDb
7482
}
7583

76-
if err := migrateDatabase(sourceConfig, targetConfig); err != nil {
84+
if err := migrateDatabase(sourceConfig, targetConfig, limitTables); err != nil {
7785
logrus.Fatalf("Migration failed: %v", err)
7886
}
7987
}
8088

81-
func migrateDatabase(source, target DbConfig) error {
89+
func migrateDatabase(source, target DbConfig, limitTables []string) error {
8290
var sourceDb *sqlx.DB
8391
var err error
8492

@@ -129,6 +137,17 @@ func migrateDatabase(source, target DbConfig) error {
129137
return fmt.Errorf("failed to get table names: %v", err)
130138
}
131139

140+
// Filter tables if limitTables is provided
141+
if len(limitTables) > 0 {
142+
filteredTables := make([]string, 0)
143+
for _, table := range tables {
144+
if slices.Contains(limitTables, table) {
145+
filteredTables = append(filteredTables, table)
146+
}
147+
}
148+
tables = filteredTables
149+
}
150+
132151
// Migrate each table
133152
for _, table := range tables {
134153
logrus.Printf("Migrating table: %s", table)

db/epochs.go

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,3 +81,37 @@ func GetEpochs(firstEpoch uint64, limit uint32) []*dbtypes.Epoch {
8181
}
8282
return epochs
8383
}
84+
85+
// EpochParticipation represents participation data for a finalized canonical epoch
86+
type EpochParticipation struct {
87+
Epoch uint64 `db:"epoch"`
88+
BlockCount uint64 `db:"block_count"`
89+
Eligible uint64 `db:"eligible"`
90+
VotedTarget uint64 `db:"voted_target"`
91+
VotedHead uint64 `db:"voted_head"`
92+
VotedTotal uint64 `db:"voted_total"`
93+
}
94+
95+
// GetFinalizedEpochParticipation gets participation data for finalized canonical epochs in the given range
96+
func GetFinalizedEpochParticipation(startEpoch, endEpoch uint64) ([]*EpochParticipation, error) {
97+
var results []*EpochParticipation
98+
99+
err := ReaderDb.Select(&results, `
100+
SELECT
101+
epoch,
102+
block_count,
103+
eligible,
104+
voted_target,
105+
voted_head,
106+
voted_total
107+
FROM epochs
108+
WHERE epoch >= $1 AND epoch <= $2
109+
ORDER BY epoch
110+
`, startEpoch, endEpoch)
111+
112+
if err != nil {
113+
return nil, err
114+
}
115+
116+
return results, nil
117+
}

db/forks.go

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,3 +151,152 @@ func GetForkById(forkId uint64) *dbtypes.Fork {
151151

152152
return &fork
153153
}
154+
155+
func GetForkVisualizationData(startSlot uint64, endSlot uint64) ([]*dbtypes.Fork, error) {
156+
forks := []*dbtypes.Fork{}
157+
158+
// Get forks that overlap with our time window and their direct parents/children
159+
err := ReaderDb.Select(&forks, `
160+
SELECT DISTINCT fork_id, base_slot, base_root, leaf_slot, leaf_root, parent_fork
161+
FROM (
162+
-- Forks that overlap with our time window
163+
SELECT fork_id, base_slot, base_root, leaf_slot, leaf_root, parent_fork
164+
FROM forks
165+
WHERE base_slot < $2 AND leaf_slot >= $1
166+
167+
UNION
168+
169+
-- Direct parents of forks in our window
170+
SELECT p.fork_id, p.base_slot, p.base_root, p.leaf_slot, p.leaf_root, p.parent_fork
171+
FROM forks p
172+
INNER JOIN forks f ON p.fork_id = f.parent_fork
173+
WHERE f.base_slot < $2 AND f.leaf_slot >= $1
174+
175+
UNION
176+
177+
-- Direct children of forks in our window
178+
SELECT c.fork_id, c.base_slot, c.base_root, c.leaf_slot, c.leaf_root, c.parent_fork
179+
FROM forks c
180+
INNER JOIN forks f ON c.parent_fork = f.fork_id
181+
WHERE f.base_slot < $2 AND f.leaf_slot >= $1
182+
) AS combined_forks
183+
ORDER BY base_slot ASC, fork_id ASC
184+
`, startSlot, endSlot)
185+
if err != nil {
186+
return nil, err
187+
}
188+
189+
return forks, nil
190+
}
191+
192+
// GetForkBlockCounts returns the number of blocks for each fork ID
193+
func GetForkBlockCounts(startSlot uint64, endSlot uint64) (map[uint64]uint64, error) {
194+
args := []any{startSlot, endSlot}
195+
196+
query := `
197+
SELECT fork_id, COUNT(*) as block_count
198+
FROM slots
199+
WHERE status != 0
200+
AND slot >= $1 AND slot < $2
201+
GROUP BY fork_id
202+
`
203+
204+
rows, err := ReaderDb.Query(query, args...)
205+
if err != nil {
206+
return nil, err
207+
}
208+
defer rows.Close()
209+
210+
blockCounts := make(map[uint64]uint64)
211+
for rows.Next() {
212+
var forkId, count uint64
213+
if err := rows.Scan(&forkId, &count); err != nil {
214+
return nil, err
215+
}
216+
blockCounts[forkId] = count
217+
}
218+
219+
return blockCounts, nil
220+
}
221+
222+
func GetSyncCommitteeParticipation(slot uint64) (float64, error) {
223+
var participation float64
224+
225+
// Handle potential overflow by using int64
226+
startSlot := int64(slot) - 1800
227+
endSlot := int64(slot) + 1800
228+
229+
// Ensure we don't go below 0
230+
if startSlot < 0 {
231+
startSlot = 0
232+
}
233+
234+
err := ReaderDb.Get(&participation, `
235+
SELECT
236+
COALESCE(AVG(CASE WHEN sync_participation IS NOT NULL
237+
THEN sync_participation ELSE 0 END), 0) as avg_participation
238+
FROM slots
239+
WHERE slot >= $1 AND slot < $2
240+
AND sync_participation IS NOT NULL
241+
`, startSlot, endSlot)
242+
if err != nil {
243+
return 0, err
244+
}
245+
246+
return participation, nil
247+
}
248+
249+
// ForkParticipationByEpoch represents participation data for a fork in a specific epoch
250+
type ForkParticipationByEpoch struct {
251+
ForkId uint64 `db:"fork_id"`
252+
Epoch uint64 `db:"epoch"`
253+
Participation float64 `db:"avg_participation"`
254+
SlotCount uint64 `db:"slot_count"`
255+
}
256+
257+
// GetForkParticipationByEpoch gets average participation per fork per epoch for the given epoch range
258+
// This is optimized to fetch all data in one query to avoid expensive repeated queries
259+
func GetForkParticipationByEpoch(startEpoch, endEpoch uint64, forkIds []uint64) ([]*ForkParticipationByEpoch, error) {
260+
if len(forkIds) == 0 {
261+
return []*ForkParticipationByEpoch{}, nil
262+
}
263+
264+
// Build IN clause for fork IDs
265+
var forkPlaceholders strings.Builder
266+
args := make([]interface{}, 0, len(forkIds)+2)
267+
268+
for i, forkId := range forkIds {
269+
if i > 0 {
270+
forkPlaceholders.WriteString(",")
271+
}
272+
args = append(args, forkId)
273+
fmt.Fprintf(&forkPlaceholders, "$%d", len(args))
274+
}
275+
276+
// Add epoch range parameters
277+
args = append(args, startEpoch, endEpoch)
278+
startEpochParam := len(args) - 1
279+
endEpochParam := len(args)
280+
281+
query := fmt.Sprintf(`
282+
SELECT
283+
s.fork_id,
284+
(s.slot / 32) AS epoch,
285+
AVG(COALESCE(s.sync_participation, 0)) AS avg_participation,
286+
COUNT(*) AS slot_count
287+
FROM slots s
288+
WHERE s.fork_id IN (%s)
289+
AND (s.slot / 32) >= $%d AND (s.slot / 32) <= $%d
290+
AND s.sync_participation IS NOT NULL
291+
GROUP BY s.fork_id, (s.slot / 32)
292+
ORDER BY s.fork_id, (s.slot / 32)
293+
`, forkPlaceholders.String(), startEpochParam, endEpochParam)
294+
295+
var results []*ForkParticipationByEpoch
296+
err := ReaderDb.Select(&results, query, args...)
297+
if err != nil {
298+
return nil, fmt.Errorf("error fetching fork participation by epoch: %w", err)
299+
}
300+
301+
return results, nil
302+
}

0 commit comments

Comments
 (0)