add discv4 ethereum execution layer support

Author: dennis-tra

cmd/nebula/cmd_crawl.go

@@ -21,6 +21,7 @@ import (
 	"github.com/dennis-tra/nebula-crawler/core"
 	"github.com/dennis-tra/nebula-crawler/db"
 	"github.com/dennis-tra/nebula-crawler/db/models"
+	"github.com/dennis-tra/nebula-crawler/discv4"
 	"github.com/dennis-tra/nebula-crawler/discv5"
 	"github.com/dennis-tra/nebula-crawler/libp2p"
 )
@@ -233,6 +234,47 @@ func CrawlAction(c *cli.Context) error {
 	}
 
 	switch cfg.Network {
+	case string(config.NetworkEthExec):
+		// configure the crawl driver
+		driverCfg := &discv4.CrawlDriverConfig{
+			Version:           cfg.Root.Version(),
+			DialTimeout:       cfg.Root.DialTimeout,
+			TrackNeighbors:    cfg.PersistNeighbors,
+			BootstrapPeerStrs: cfg.BootstrapPeers.Value(),
+			AddrDialType:      cfg.AddrDialType(),
+			AddrTrackType:     cfg.AddrTrackType(),
+			TracerProvider:    cfg.Root.TracerProvider,
+			MeterProvider:     cfg.Root.MeterProvider,
+			LogErrors:         cfg.Root.LogErrors,
+		}
+
+		// init the crawl driver
+		driver, err := discv4.NewCrawlDriver(dbc, dbCrawl, driverCfg)
+		if err != nil {
+			return fmt.Errorf("new discv4 driver: %w", err)
+		}
+
+		// init the result handler
+		handler := core.NewCrawlHandler[discv4.PeerInfo](handlerCfg)
+
+		// put everything together and init the engine that'll run the crawl
+		eng, err := core.NewEngine[discv4.PeerInfo, core.CrawlResult[discv4.PeerInfo]](driver, handler, engineCfg)
+		if err != nil {
+			return fmt.Errorf("new engine: %w", err)
+		}
+
+		// finally, start the crawl
+		queuedPeers, runErr := eng.Run(ctx)
+
+		// a bit ugly but, but the handler will contain crawl statistics, that
+		// we'll save to the database and print to the screen
+		handler.QueuedPeers = len(queuedPeers)
+		if err := persistCrawlInformation(dbc, dbCrawl, handler, runErr); err != nil {
+			return fmt.Errorf("persist crawl information: %w", err)
+		}
+
+		return nil
+
 	case string(config.NetworkEthCons),
 		string(config.NetworkHolesky): // use a different driver etc. for the Ethereum consensus layer + Holeksy Testnet
 		// configure the crawl driver
@@ -252,7 +294,7 @@ func CrawlAction(c *cli.Context) error {
 		// init the crawl driver
 		driver, err := discv5.NewCrawlDriver(dbc, dbCrawl, driverCfg)
 		if err != nil {
-			return fmt.Errorf("new driver: %w", err)
+			return fmt.Errorf("new discv5 driver: %w", err)
 		}
 
 		// init the result handler

config/bootstrap.go

@@ -201,6 +201,15 @@ var (
 		"enr:-LK4QKWrXTpV9T78hNG6s8AM6IO4XH9kFT91uZtFg1GcsJ6dKovDOr1jtAAFPnS2lvNltkOGA9k29BUN7lFh_sjuc9QBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhANAdd-Jc2VjcDI1NmsxoQLQa6ai7y9PMN5hpLe5HmiJSlYzMuzP7ZhwRiwHvqNXdoN0Y3CCI4yDdWRwgiOM",
 	}
 
+	// BootstrapPeersEthereumExecution extracted from:
+	//   https://github.com/ethereum/go-ethereum/blob/f04e5bde7487ce554930187e766164b18c37d867/params/bootnodes.go#L23
+	BootstrapPeersEthereumExecution = []string{
+		"enode://d860a01f9722d78051619d1e2351aba3f43f943f6f00718d1b9baa4101932a1f5011f16bb2b1bb35db20d6fe28fa0bf09636d26a87d31de9ec6203eeedb1f666@18.138.108.67:30303", // bootnode-aws-ap-southeast-1-001
+		//"enode://22a8232c3abc76a16ae9d6c3b164f98775fe226f0917b0ca871128a74a8e9630b458460865bab457221f1d448dd9791d24c4e5d88786180ac185df813a68d4de@3.209.45.79:30303",   // bootnode-aws-us-east-1-001
+		//"enode://2b252ab6a1d0f971d9722cb839a42cb81db019ba44c08754628ab4a823487071b5695317c8ccd085219c3a03af063495b2f1da8d18218da2d6a82981b45e6ffc@65.108.70.101:30303", // bootnode-hetzner-hel
+		//"enode://4aeb4ab6c14b23e2c4cfdce879c04b0748a20d8e9b59e25ded2a08143e265c6c25936e74cbc8e641e3312ca288673d91f2f93f8e277de3cfa444ecdaaf982052@157.90.35.166:30303", // bootnode-hetzner-fsn
+	}
+
 	// BootstrapPeersHolesky extracted from:
 	//   https://github.com/eth-clients/holesky/blob/main/custom_config_data/bootstrap_nodes.txt
 	BootstrapPeersHolesky = []string{

config/config.go

@@ -28,6 +28,7 @@ const (
 	NetworkMocha    Network = "MOCHA"
 	NetworkBlockRa  Network = "BLOCKSPACE_RACE"
 	NetworkEthCons  Network = "ETHEREUM_CONSENSUS"
+	NetworkEthExec  Network = "ETHEREUM_EXECUTION"
 	NetworkHolesky  Network = "HOLESKY"
 )
 
@@ -45,6 +46,7 @@ func Networks() []Network {
 		NetworkMocha,
 		NetworkBlockRa,
 		NetworkEthCons,
+		NetworkEthExec,
 		NetworkHolesky,
 	}
 }
@@ -285,6 +287,9 @@ func ConfigureNetwork(network string) (*cli.StringSlice, *cli.StringSlice, error
 	case NetworkEthCons:
 		bootstrapPeers = cli.NewStringSlice(BootstrapPeersEthereumConsensus...)
 		protocols = cli.NewStringSlice("discv5") // TODO
+	case NetworkEthExec:
+		bootstrapPeers = cli.NewStringSlice(BootstrapPeersEthereumExecution...)
+		protocols = cli.NewStringSlice("discv4") // TODO
 	case NetworkHolesky:
 		bootstrapPeers = cli.NewStringSlice(BootstrapPeersHolesky...)
 		protocols = cli.NewStringSlice("discv5") // TODO

discv4/crawler.go

@@ -0,0 +1,208 @@
+package discv4
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	log "github.com/sirupsen/logrus"
+	"go.uber.org/atomic"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/dennis-tra/nebula-crawler/config"
+	"github.com/dennis-tra/nebula-crawler/core"
+	"github.com/dennis-tra/nebula-crawler/db"
+	"github.com/dennis-tra/nebula-crawler/discvx"
+)
+
+type CrawlerConfig struct {
+	DialTimeout  time.Duration
+	AddrDialType config.AddrType
+	LogErrors    bool
+}
+
+type Crawler struct {
+	id           string
+	cfg          *CrawlerConfig
+	listener     *discvx.UDPv4
+	crawledPeers int
+	done         chan struct{}
+}
+
+var _ core.Worker[PeerInfo, core.CrawlResult[PeerInfo]] = (*Crawler)(nil)
+
+func (c *Crawler) Work(ctx context.Context, task PeerInfo) (core.CrawlResult[PeerInfo], error) {
+	logEntry := log.WithFields(log.Fields{
+		"crawlerID":  c.id,
+		"remoteID":   task.peerID.ShortString(),
+		"crawlCount": c.crawledPeers,
+	})
+	logEntry.Debugln("Crawling peer")
+	defer logEntry.Debugln("Crawled peer")
+
+	crawlStart := time.Now()
+
+	result := c.crawlDiscV4(ctx, task)
+
+	cr := core.CrawlResult[PeerInfo]{
+		CrawlerID:           c.id,
+		Info:                task,
+		CrawlStartTime:      crawlStart,
+		RoutingTableFromAPI: false,
+		RoutingTable:        result.RoutingTable,
+		// Agent:               libp2pResult.Agent,
+		// Protocols:           libp2pResult.Protocols,
+		// ConnectError:        libp2pResult.ConnectError,
+		// ConnectErrorStr:     libp2pResult.ConnectErrorStr,
+		CrawlError:    result.Error,
+		CrawlErrorStr: result.ErrorStr,
+		CrawlEndTime:  time.Now(),
+		// ConnectStartTime:    libp2pResult.ConnectStartTime,
+		// ConnectEndTime:      libp2pResult.ConnectEndTime,
+		// Properties: data,
+		LogErrors: c.cfg.LogErrors,
+	}
+
+	// We've now crawled this peer, so increment
+	c.crawledPeers++
+
+	return cr, nil
+}
+
+type DiscV4Result struct {
+	// The time we received the first successful response
+	RespondedAt *time.Time
+
+	// The updated ethereum node record
+	ENR *enode.Node
+
+	// The neighbors of the crawled peer
+	RoutingTable *core.RoutingTable[PeerInfo]
+
+	// The time the draining of bucket entries was finished
+	DoneAt time.Time
+
+	// The combined error of crawling the peer's buckets
+	Error error
+
+	// The above error mapped to a known string
+	ErrorStr string
+}
+
+func (c *Crawler) crawlDiscV4(ctx context.Context, pi PeerInfo) DiscV4Result {
+	// mutex to guard access to result and allNeighbors
+	mu := sync.RWMutex{}
+
+	// the final result struct
+	result := DiscV4Result{}
+
+	// all neighbors of pi. We're using a map to deduplicate.
+	allNeighbors := map[string]PeerInfo{}
+
+	// errorBits tracks at which CPL errors have occurred.
+	// 0000 0000 0000 0000 - No error
+	// 0000 0000 0000 0001 - An error has occurred at CPL 0
+	// 1000 0000 0000 0001 - An error has occurred at CPL 0 and 15
+	errorBits := atomic.NewUint32(0)
+
+	enr, err := c.listener.RequestENR(pi.Node)
+	if err != nil {
+		result.ENR = pi.Node
+	} else {
+		result.ENR = enr
+		now := time.Now()
+		result.RespondedAt = &now
+	}
+
+	errg := errgroup.Group{}
+	for i := 0; i <= 15; i++ { // 15 is maximum
+		count := i // Copy value
+		errg.Go(func() error {
+			pubKey, err := discvx.GenRandomPublicKey(pi.Node.ID(), count)
+			if err != nil {
+				log.WithError(err).WithField("enr", pi.Node.String()).Warnln("Failed generating public key")
+				errorBits.Add(1 << count)
+				return fmt.Errorf("generating random public key with CPL %d: %w", count, err)
+			}
+
+			udpAddr := &net.UDPAddr{IP: pi.Node.IP(), Port: pi.Node.UDP()}
+
+			var neighbors []*enode.Node
+			for retry := 0; retry < 2; retry++ {
+				neighbors, err = c.listener.FindNode(pi.Node.ID(), udpAddr, pubKey)
+				if err == nil {
+					break
+				}
+
+				errorBits.Add(1 << count)
+
+				if errors.Is(err, discvx.ErrTimeout) {
+					sleepDur := time.Second * time.Duration(5*(retry+1))
+					select {
+					case <-ctx.Done():
+						return ctx.Err()
+					case <-time.After(sleepDur): // may add jitter here
+						continue
+					}
+				}
+
+				errorBits.Add(1 << count)
+
+				return fmt.Errorf("getting closest peer with CPL %d: %w", count, err)
+			}
+
+			mu.Lock()
+			defer mu.Unlock()
+
+			if result.RespondedAt == nil {
+				now := time.Now()
+				result.RespondedAt = &now
+			}
+
+			for _, n := range neighbors {
+				npi, err := NewPeerInfo(n)
+				if err != nil {
+					log.WithError(err).Warnln("Failed parsing ethereum node neighbor")
+					continue
+				}
+				allNeighbors[string(npi.peerID)] = npi
+			}
+
+			if err != nil {
+				errorBits.Add(1 << count)
+				return err
+			}
+
+			return nil
+		})
+	}
+
+	// wait for go routines to finish
+	err = errg.Wait()
+
+	// track done timestamp and error
+	result.DoneAt = time.Now()
+	result.Error = err
+
+	result.RoutingTable = &core.RoutingTable[PeerInfo]{
+		PeerID:    pi.ID(),
+		Neighbors: []PeerInfo{},
+		ErrorBits: uint16(errorBits.Load()),
+		Error:     err,
+	}
+
+	for _, n := range allNeighbors {
+		result.RoutingTable.Neighbors = append(result.RoutingTable.Neighbors, n)
+	}
+
+	// if there was a connection error, parse it to a known one
+	if result.Error != nil {
+		result.ErrorStr = db.NetError(result.Error)
+	}
+
+	return result
+}

discv4/dialer.go

@@ -0,0 +1,56 @@
+package discv4
+
+import (
+	"context"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/dennis-tra/nebula-crawler/core"
+	"github.com/dennis-tra/nebula-crawler/db"
+	"github.com/dennis-tra/nebula-crawler/discvx"
+)
+
+// Dialer encapsulates a libp2p host that dials peers.
+type Dialer struct {
+	id          string
+	dialedPeers uint64
+	listener    *discvx.UDPv4
+}
+
+var _ core.Worker[PeerInfo, core.DialResult[PeerInfo]] = (*Dialer)(nil)
+
+// Work takes the PeerInfo object and tries to figure out if the peer is
+// still online.
+func (d *Dialer) Work(ctx context.Context, task PeerInfo) (core.DialResult[PeerInfo], error) {
+	// Creating log entry
+	logEntry := log.WithFields(log.Fields{
+		"dialerID":  d.id,
+		"remoteID":  task.ID().ShortString(),
+		"dialCount": d.dialedPeers,
+	})
+	logEntry.Debugln("Dialing peer")
+	defer logEntry.Debugln("Dialed peer")
+
+	// Initialize dial result
+	dr := core.DialResult[PeerInfo]{
+		DialerID:      d.id,
+		Info:          task,
+		DialStartTime: time.Now(),
+	}
+
+	newEnr, err := d.listener.RequestENR(task.Node)
+	dr.DialEndTime = time.Now()
+
+	if err != nil {
+		dr.Error = err
+		dr.DialError = db.NetError(dr.Error)
+	} else {
+		task.Node = newEnr
+		dr.Info = task
+	}
+
+	d.dialedPeers += 1
+
+	return dr, nil
+}