Merge upstream migalabs/goteth 3.9.0 into Obol fork

.github/workflows/ci.yml

@@ -0,0 +1,39 @@
+name: CI
+
+on:
+  push:
+    branches: [main, master, dev]
+  pull_request:
+
+# The job only reads the code: give its token read access, like build_push.yml.
+permissions:
+  contents: read
+
+# This repository had no CI at all. Every change to the indexer serving
+# mainnet merged without being compiled, vetted or tested by anything but
+# whoever wrote it.
+jobs:
+  build-and-test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          # go.mod replaces go-relay-client with a local path.
+          submodules: recursive
+
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+          cache: true
+
+      - name: Build
+        run: go build ./...
+
+      - name: Vet
+        run: go vet ./...
+
+      - name: Test
+        # The metrics tests need a live beacon node and skip themselves when
+        # none answers; set GOTETH_TEST_BN_ENDPOINT (and GOTETH_TEST_EL_ENDPOINT)
+        # to run them against one.
+        run: go test -race ./...

CITATION.cff

@@ -0,0 +1,31 @@
+cff-version: 1.2.0
+message: "If you use this software, please cite it using the metadata from this file."
+type: software
+title: GotEth
+abstract: >-
+  GotEth is a go-written client that indexes all validator-related duties,
+  parameters and transactions from Ethereum's consensus and execution layers.
+authors:
+  - given-names: Tarun
+    family-names: Mohandas
+    affiliation: MigaLabs
+  - alias: Zyra-V21
+    given-names: Ruben
+    family-names: Planes Llinares
+    affiliation: MigaLabs
+  - given-names: Leo
+    family-names: Bautista-Gomez
+    affiliation: MigaLabs
+  - given-names: Mikel
+    family-names: Cortes
+    affiliation: MigaLabs
+repository-code: "https://github.com/migalabs/goteth"
+license: MIT
+keywords:
+  - ethereum
+  - consensus layer
+  - beacon chain
+  - validators
+  - clickhouse
+version: v3.9.0
+date-released: "2026-10-22"

README.md

@@ -138,9 +138,15 @@ If specific upgrades or downgrades need to be done manually, one could do this w
 During `v3.0.0` we migrated our database system from PostgreSQL to Clickhouse.
 If you wish to migrate your existing database, please follow [this](https://migalabs.notion.site/PostgreSQL-to-Clickhouse-migration-611a52a457824cd494d701773365f62f) guide.
 
+# Citation
+
+[![DOI](https://zenodo.org/badge/DOI/10.5281/zenodo.XXXXXXX.svg)](https://doi.org/10.5281/zenodo.XXXXXXX)
+
+If you use GotEth in your work, please cite it using the metadata in [CITATION.cff](CITATION.cff), or the "Cite this repository" button on GitHub.
+
 # Maintainers
 
-@santi1234567
+@Zyra-V21
 
 # Contributing
 

cmd/blocks_cmd.go

@@ -99,7 +99,7 @@ var BlocksCommand = &cli.Command{
 		},
 		&cli.StringFlag{
 			Name:        "metrics",
-			Usage:       "Metrics to be persisted to the database: epoch,block,rewards,transactions,api_rewards,blob_sidecars",
+			Usage:       "Metrics to be persisted to the database: epoch,block,rewards,transactions,api_rewards,blob_sidecars. blob_sidecars covers blob metadata (t_blob_sidecars); blob p2p arrival timings are always recorded in head mode via the data_column_sidecar event into t_data_column_sidecars_events (the post-Fulu replacement of the dead blob_sidecar event)",
 			EnvVars:     []string{"ANALYZER_METRICS"},
 			DefaultText: "epoch,block",
 		},

cmd/pool_summary_refill_cmd.go

@@ -0,0 +1,71 @@
+package cmd
+
+import (
+	"fmt"
+
+	"github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/migalabs/goteth/pkg/db"
+	"github.com/migalabs/goteth/pkg/utils"
+	"github.com/sirupsen/logrus"
+	cli "github.com/urfave/cli/v2"
+)
+
+var PoolSummaryRefillCommand = &cli.Command{
+	Name:   "pool-summary-refill",
+	Usage:  "Recomputes t_pool_summary for an epoch range from the current rewards and pool tags. Deletes the range first so pool renames do not leave ghost rows. Use after pool re-tagging or rewards backfills.",
+	Action: LaunchPoolSummaryRefill,
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:        "log-level",
+			Usage:       "Log level: debug, warn, info, error",
+			EnvVars:     []string{"ANALYZER_LOG_LEVEL"},
+			DefaultText: "info",
+		},
+		&cli.StringFlag{
+			Name:        "db-url",
+			Usage:       "Database where the pool summaries are persisted",
+			EnvVars:     []string{"ANALYZER_DB_URL"},
+			DefaultText: "clickhouse://username:password@localhost:9000/goteth?x-multi-statement=true&max_memory_usage=10000000000",
+		},
+		&cli.Uint64Flag{
+			Name:     "from-epoch",
+			Usage:    "First epoch of the range to recompute (inclusive)",
+			Required: true,
+		},
+		&cli.Uint64Flag{
+			Name:     "to-epoch",
+			Usage:    "Last epoch of the range to recompute (inclusive)",
+			Required: true,
+		},
+	},
+}
+
+func LaunchPoolSummaryRefill(c *cli.Context) error {
+
+	if c.IsSet("log-level") {
+		logrus.SetLevel(utils.ParseLogLevel(c.String("log-level")))
+	}
+
+	dbUrl := c.String("db-url")
+	if dbUrl == "" {
+		return fmt.Errorf("db-url is required")
+	}
+
+	fromEpoch := phase0.Epoch(c.Uint64("from-epoch"))
+	toEpoch := phase0.Epoch(c.Uint64("to-epoch"))
+	if toEpoch < fromEpoch {
+		return fmt.Errorf("to-epoch (%d) must be greater or equal than from-epoch (%d)", toEpoch, fromEpoch)
+	}
+
+	dbClient, err := db.New(c.Context, dbUrl)
+	if err != nil {
+		return err
+	}
+	err = dbClient.Connect()
+	if err != nil {
+		return err
+	}
+	defer dbClient.Finish()
+
+	return dbClient.RefreshPoolSummaryRange(fromEpoch, toEpoch)
+}

cmd/validator_window_cmd.go

@@ -43,6 +43,12 @@ var ValidatorWindowCommand = &cli.Command{
 			EnvVars:     []string{"DELETE_CADENCE_EPOCHS"},
 			DefaultText: "32",
 		},
+		&cli.IntFlag{
+			Name:        "pool-reroll-epochs",
+			Usage:       "Refresh t_pool_summary for the last N epochs whenever the validator->pool mapping in t_eth2_pubkeys changes, so late-tagged validators and pool renames propagate to recent history. Set to 0 to disable.",
+			EnvVars:     []string{"POOL_REROLL_EPOCHS"},
+			DefaultText: "0",
+		},
 		&cli.StringFlag{
 			Name:        "bn-endpoint",
 			Usage:       "Beacon node endpoint (to request the Beacon States and Blocks)",

docs/tables.md

@@ -156,12 +156,32 @@ Config: `engine = MergeTree ORDER BY f_val_idx`
 | f_status                 | uint8        | status (see status table)                           |
 | f_slashed                | bool         | whether the validator has ever been slashed or not  |
 | f_activation_epoch       | uint64       | epoch at which the validator was activated          |
+| f_activation_eligibility_epoch | uint64 | epoch at which the validator became eligible for activation; `FAR_FUTURE_EPOCH` (`18446744073709551615`) means not yet eligible. Combined with `f_activation_epoch`, this splits the pre-activation queue (`f_status = 0`) into the two deposit-lifecycle states described below. |
 | f_withdrawal_epoch       | uint64       | epoch at which the validator can withdraw funds     |
 | f_exit_epoch             | uint64       | epoch at which the validator exited the network     |
 | f_public_key             | string       | public key of the validator                         |
 | f_withdrawal_prefix      | uint8        | withdrawal prefix of the validator's credentials    |
 | f_withdrawal_credentials | text         | withdrawal credentials of the validator (see below) |
 
+## Appendix: deposit-lifecycle status vocabulary
+
+`f_status` keeps its coarse, stable encoding (`0 in_activation_queue · 1 active · 2 exited · 3 slashed`).
+Consumers that want the finer deposit-lifecycle vocabulary derive it from
+`f_status` + `f_activation_eligibility_epoch` (and, for pre-registration deposits, from
+`t_deposit_requests`). This is a proposed vocabulary the current Beacon API `ValidatorStatus`
+enum does not provide — its `pending_initialized` / `pending_queued` predate the post-Electra
+deposit pipeline (EIP-6110/EIP-7251) and there is no deposit-queue state:
+
+| Derived status | Condition | Meaning |
+| -------------- | --------- | ------- |
+| `deposit_inqueue` | pubkey present in `t_deposit_requests`, absent from `t_validator_last_status` (not yet a registered validator) | Deposit waiting in `state.pending_deposits`; the real churn-limited queue |
+| `pending_minBalance` | `f_status = 0` and `f_activation_eligibility_epoch = FAR_FUTURE_EPOCH` | Validator created, effective balance < 32 ETH (not yet eligible) |
+| `pending_lookahead` | `f_status = 0` and `f_activation_eligibility_epoch != FAR_FUTURE_EPOCH` | Eligible; fixed `1 + MAX_SEED_LOOKAHEAD` delay (not a queue) |
+| `active` / `exited` / `slashed` | `f_status` 1 / 2 / 3 | unchanged |
+
+`deposit_inqueue` is not a `t_validator_last_status` row because the deposit has no validator index
+yet; it is materialised by downstream consumers (e.g. bindeth) joining `t_deposit_requests`.
+
 ## Appendix: Reference for `f_withdrawal_prefix`
 
 The `f_withdrawal_prefix` column indicates the type of withdrawal credentials associated with the validator. The possible values are:
@@ -307,18 +327,24 @@ Config: `engine = ReplacingMergeTree ORDER BY f_slot, f_index`
 | f_kzg_proof      | string       | kzg proof of the blob                                                                                                         |
 | f_ending_0s      | uint64       | amount of consecutive 0s at the end of the blob bytes                                                                         |
 
-# Blob Sidecars Events (`t_blob_sidecars_events`)
+# Data Column Sidecars Events (`t_data_column_sidecars_events`)
+
+Config: `engine = ReplacingMergeTree PARTITION BY intDiv(f_slot, 100000) ORDER BY (f_slot, f_column_index, f_block_root)`
+
+P2P arrival timings for blob data. Since the Fulu fork (PeerDAS) blobs are no longer gossiped whole: they are erasure-coded into columns and each node custodies a subset, so beacon nodes emit `data_column_sidecar` instead of the now-dead `blob_sidecar` event. One row per **column** arrival (up to 128 per slot), which also exposes the reconstruction curve of each blob rather than a single arrival timestamp.
+
+Blob *metadata* is unaffected and still comes from the block itself (`t_blob_sidecars`).
 
-Config: `engine = ReplacingMergeTree ORDER BY f_arrival_timestamp_ms, f_blob_hash, f_slot`
+| Column Name            | Type of Data  | Description                                                    |     |     |
+| ---------------------- | ------------- | -------------------------------------------------------------- | --- | --- |
+| f_arrival_timestamp_ms | uint64        | timestamp at which goteth received the data column event       |
+| f_slot                 | uint64        | slot the data column belongs to                                |
+| f_block_root           | string        | block root hash                                                |
+| f_column_index         | uint16        | index of the data column (0..127)                              |
+| f_kzg_commitments      | array(string) | kzg commitments of the blobs this column carries a piece of    |
+| f_num_commitments      | uint16        | number of commitments (denormalised from the array for queries)|
 
-| Column Name            | Type of Data | Description                                       |     |     |
-| ---------------------- | ------------ | ------------------------------------------------- | --- | --- |
-| f_arrival_timestamp_ms | uint64       | timestamp at which goteth received the blob event |
-| f_blob_hash            | string       | hash of the blob                                  |
-| f_slot                 | uint64       | slot at which the blob was sent                   |
-| f_block_root           | string       | block root hash                                   |
-| f_index                | uint8        | index of the blob                                 |
-| f_kzg_commitment       | string       | kzg commitment of the blob                        |
+> `t_blob_sidecars_events` is the pre-Fulu predecessor of this table (one row per whole blob). It can no longer receive rows because the `blob_sidecar` event is never emitted post-Fulu; migration `000038` renames it to `t_blob_sidecars_events_pre_fulu` to preserve its history.
 
 # Block Rewards (`t_block_rewards`)
 

main.go

@@ -47,6 +47,7 @@ func main() {
 		Commands: []*cli.Command{
 			cmd.BlocksCommand,
 			cmd.ValidatorWindowCommand,
+			cmd.PoolSummaryRefillCommand,
 		},
 	}
 

pkg/analyzer/chain_analyzer.go

@@ -3,6 +3,7 @@ package analyzer
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/attestantio/go-eth2-client/spec/phase0"
@@ -42,7 +43,7 @@ type ChainAnalyzer struct {
 	// Control Variables
 	wgMainRoutine            *sync.WaitGroup    // wait group for main routine (either historical or head)
 	wgDownload               *sync.WaitGroup    // wait group for download routine
-	stop                     bool               // flag to notify all routine to finish
+	stop                     atomic.Bool        // flag to notify all routine to finish (read and written from several goroutines)
 	routineClosed            chan struct{}      // signal that everything was closed succesfully
 	downloadMode             string             // whether to download historical blocks (defined by user) or follow chain head
 	rewardsAggregationEpochs int                // number of epochs to aggregate rewards
@@ -51,12 +52,15 @@ type ChainAnalyzer struct {
 	metrics                  db.DBMetrics       // what metrics to be downloaded / processed
 	processerBook            *utils.RoutineBook // defines slot to process new metrics into the database, good for monitoring
 
-	downloadCache                    ChainCache // store the blocks and states downloaded
+	downloadCache                   ChainCache // store the blocks and states downloaded
 	validatorsRewardsAggregations   map[phase0.ValidatorIndex]*spec.ValidatorRewardsAggregation
 	validatorsRewardsAggregationsMu sync.Mutex
-	advanceFinalizedMu              sync.Mutex // serializes AdvanceFinalized so concurrent invocations cannot race CleanUpTo against in-flight Wait()s
+	advanceFinalizedMu              sync.Mutex            // serializes AdvanceFinalized so concurrent invocations cannot race CleanUpTo against in-flight Wait()s
+	handleReorgMu                   sync.Mutex            // serializes HandleReorg so two walks cannot process the same epoch at once and leak processerBook slots (#292)
+	pendingReprocess                map[uint64]bool       // epochs whose derived rows went stale when an earlier epoch was rewritten, and that AdvanceFinalized could not reach in that invocation (#285). Guarded by advanceFinalizedMu.
+	reprocessAttempts               map[uint64]int        // how many times each pending epoch has been attempted, so a debt that cannot be paid is abandoned loudly rather than retried forever (#291). Guarded by advanceFinalizedMu.
 	aggregatedEpochsInWindow        map[phase0.Epoch]bool // set of unique epochs aggregated in current window; prevents double-counting on reprocessing (#255)
-	epochBoundaryStateRoots       sync.Map   // slot -> phase0.Root, caches state roots from Head SSE events at epoch boundaries
+	epochBoundaryStateRoots         sync.Map              // slot -> phase0.Root, caches state roots from Head SSE events at epoch boundaries
 
 	initTime    time.Time
 	PromMetrics *prom_metrics.PrometheusMetrics // metrics to be stored to prometheus
@@ -182,6 +186,7 @@ func NewChainAnalyzer(
 		downloadCache:                 NewQueue(),
 		validatorsRewardsAggregations: make(map[phase0.ValidatorIndex]*spec.ValidatorRewardsAggregation),
 		aggregatedEpochsInWindow:      make(map[phase0.Epoch]bool),
+		pendingReprocess:              make(map[uint64]bool),
 		processerBook:                 utils.NewRoutineBook(32, "processer"), // one whole epoch
 		wgMainRoutine:                 &sync.WaitGroup{},
 		wgDownload:                    &sync.WaitGroup{},
@@ -218,12 +223,16 @@ func (s *ChainAnalyzer) Run() {
 		// Block requester in finalized slots, not used for now
 		s.wgMainRoutine.Add(1)
 		go s.runHead()
+
+		// Persists data_column_sidecar arrivals in batches, off the head loop (#282).
+		s.wgMainRoutine.Add(1)
+		go s.runDataColumnSidecarsPersister()
 	}
 
 	s.PromMetrics.Start()
 
 	s.wgMainRoutine.Wait()
-	s.stop = true
+	s.stop.Store(true)
 	log.Infof("main routine finished, waiting for downloader...")
 
 	s.wgDownload.Wait()
@@ -241,7 +250,7 @@ func (s *ChainAnalyzer) Run() {
 
 func (s *ChainAnalyzer) Close() {
 	log.Info("Sudden closed detected, closing StateAnalyzer")
-	s.stop = true
+	s.stop.Store(true)
 	s.cancel()
 	<-s.routineClosed // Wait for services to stop before returning
 }

pkg/analyzer/download.go

@@ -22,7 +22,7 @@ func (s *ChainAnalyzer) DownloadBlock(slot phase0.Slot) {
 	newBlock, err := s.cli.RequestBeaconBlock(slot)
 	if err != nil {
 		log.Errorf("block error at slot %d: %s", slot, err)
-		s.stop = true
+		s.stop.Store(true)
 		s.cancel()
 		return
 	}
@@ -63,7 +63,7 @@ func (s *ChainAnalyzer) DownloadState(slot phase0.Slot) {
 
 	if err != nil {
 		log.Errorf("unable to retrieve beacon state from the beacon node, closing requester routine. %s", err.Error())
-		s.stop = true
+		s.stop.Store(true)
 		s.cancel()
 		return
 	}