Merge pull request #281 from onflow/gregor/index-tx-traces

Ingest and index transaction traces

Makefile

@@ -15,13 +15,19 @@ check-tidy:
 	go mod tidy
 	git diff --exit-code
 
+.PHONY: fix-lint
+fix-lint:
+	golangci-lint run -v --fix ./...
+
 .PHONY: generate
 generate:
 	go get -d github.com/vektra/mockery/[email protected]
 	mockery --dir=storage --name=BlockIndexer --output=storage/mocks
 	mockery --dir=storage --name=ReceiptIndexer --output=storage/mocks
 	mockery --dir=storage --name=TransactionIndexer --output=storage/mocks
 	mockery --dir=storage --name=AccountIndexer --output=storage/mocks
+	mockery --dir=storage --name=TraceIndexer --output=storage/mocks
+	mockery --all --dir=services/traces --output=services/traces/mocks
 	mockery --all --dir=services/ingestion --output=services/ingestion/mocks
 	mockery --dir=models --name=Engine --output=models/mocks
 

bootstrap/bootstrap.go

@@ -19,6 +19,7 @@ import (
 	"github.com/onflow/flow-evm-gateway/models"
 	"github.com/onflow/flow-evm-gateway/services/ingestion"
 	"github.com/onflow/flow-evm-gateway/services/requester"
+	"github.com/onflow/flow-evm-gateway/services/traces"
 	"github.com/onflow/flow-evm-gateway/storage"
 	storageErrs "github.com/onflow/flow-evm-gateway/storage/errors"
 	"github.com/onflow/flow-evm-gateway/storage/pebble"
@@ -38,19 +39,12 @@ func Start(ctx context.Context, cfg *config.Config) error {
 	transactions := pebble.NewTransactions(pebbleDB)
 	receipts := pebble.NewReceipts(pebbleDB)
 	accounts := pebble.NewAccounts(pebbleDB)
+	trace := pebble.NewTraces(pebbleDB)
 
 	blocksBroadcaster := broadcast.NewBroadcaster()
 	transactionsBroadcaster := broadcast.NewBroadcaster()
 	logsBroadcaster := broadcast.NewBroadcaster()
 
-	// if database is not initialized require init height
-	if _, err := blocks.LatestCadenceHeight(); errors.Is(err, storageErrs.ErrNotInitialized) {
-		if err := blocks.InitHeights(cfg.InitCadenceHeight); err != nil {
-			return fmt.Errorf("failed to init the database: %w", err)
-		}
-		logger.Info().Msg("database initialized with 0 evm and cadence heights")
-	}
-
 	// this should only be used locally or for testing
 	if cfg.ForceStartCadenceHeight != 0 {
 		logger.Warn().Uint64("height", cfg.ForceStartCadenceHeight).Msg("force setting starting Cadence height!!!")
@@ -86,6 +80,20 @@ func Start(ctx context.Context, cfg *config.Config) error {
 		return err
 	}
 
+	// if database is not initialized require init height
+	if _, err := blocks.LatestCadenceHeight(); errors.Is(err, storageErrs.ErrNotInitialized) {
+		cadenceHeight := cfg.InitCadenceHeight
+		cadenceBlock, err := client.GetBlockHeaderByHeight(ctx, cadenceHeight)
+		if err != nil {
+			return err
+		}
+
+		if err := blocks.InitHeights(cadenceHeight, cadenceBlock.ID); err != nil {
+			return fmt.Errorf("failed to init the database: %w", err)
+		}
+		logger.Info().Msg("database initialized with 0 evm and cadence heights")
+	}
+
 	go func() {
 		err := startServer(
 			ctx,
@@ -114,6 +122,7 @@ func Start(ctx context.Context, cfg *config.Config) error {
 		transactions,
 		receipts,
 		accounts,
+		trace,
 		blocksBroadcaster,
 		transactionsBroadcaster,
 		logsBroadcaster,
@@ -134,6 +143,7 @@ func startIngestion(
 	transactions storage.TransactionIndexer,
 	receipts storage.ReceiptIndexer,
 	accounts storage.AccountIndexer,
+	trace storage.TraceIndexer,
 	blocksBroadcaster *broadcast.Broadcaster,
 	transactionsBroadcaster *broadcast.Broadcaster,
 	logsBroadcaster *broadcast.Broadcaster,
@@ -163,8 +173,44 @@ func startIngestion(
 		Uint64("missed-heights", blk.Height-latestCadenceHeight).
 		Msg("indexing cadence height information")
 
-	subscriber := ingestion.NewRPCSubscriber(client, cfg.HeartbeatInterval, cfg.FlowNetworkID, logger)
-	engine := ingestion.NewEventIngestionEngine(
+	subscriber := ingestion.NewRPCSubscriber(
+		client,
+		cfg.HeartbeatInterval,
+		cfg.FlowNetworkID,
+		logger,
+	)
+
+	if cfg.TracesEnabled {
+		downloader, err := traces.NewGCPDownloader(cfg.TracesBucketName, logger)
+		if err != nil {
+			return err
+		}
+
+		initHeight, err := blocks.LatestEVMHeight()
+		if err != nil {
+			return err
+		}
+		tracesEngine := traces.NewTracesIngestionEngine(
+			initHeight,
+			blocksBroadcaster,
+			blocks,
+			trace,
+			downloader,
+			logger,
+		)
+
+		go func() {
+			err = tracesEngine.Run(ctx)
+			if err != nil {
+				logger.Error().Err(err).Msg("traces ingestion engine failed to run")
+				panic(err)
+			}
+		}()
+
+		<-tracesEngine.Ready()
+	}
+
+	eventEngine := ingestion.NewEventIngestionEngine(
 		subscriber,
 		blocks,
 		receipts,
@@ -176,17 +222,18 @@ func startIngestion(
 		logger,
 	)
 	const retries = 15
-	restartable := models.NewRestartableEngine(engine, retries, logger)
+	restartableEventEngine := models.NewRestartableEngine(eventEngine, retries, logger)
 
 	go func() {
-		err = restartable.Run(ctx)
+		err = restartableEventEngine.Run(ctx)
 		if err != nil {
-			logger.Error().Err(err).Msg("ingestion engine failed to run")
+			logger.Error().Err(err).Msg("event ingestion engine failed to run")
 			panic(err)
 		}
 	}()
 
-	<-restartable.Ready() // wait for engine to be ready
+	// wait for ingestion engines to be ready
+	<-restartableEventEngine.Ready()
 
 	logger.Info().Msg("ingestion start up successful")
 	return nil

config/config.go

@@ -78,6 +78,10 @@ type Config struct {
 	ForceStartCadenceHeight uint64
 	// HeartbeatInterval sets custom heartbeat interval for events
 	HeartbeatInterval uint64
+	// TracesBucketName sets the GCP bucket name where transaction traces are being stored.
+	TracesBucketName string
+	// TracesEnabled sets whether the node is supporting transaction traces.
+	TracesEnabled bool
 }
 
 func FromFlags() (*Config, error) {
@@ -110,6 +114,7 @@ func FromFlags() (*Config, error) {
 	flag.IntVar(&streamTimeout, "stream-timeout", 3, "Defines the timeout in seconds the server waits for the event to be sent to the client")
 	flag.Uint64Var(&forceStartHeight, "force-start-height", 0, "Force set starting Cadence height. This should only be used locally or for testing, never in production.")
 	flag.StringVar(&filterExpiry, "filter-expiry", "5m", "Filter defines the time it takes for an idle filter to expire")
+	flag.StringVar(&cfg.TracesBucketName, "traces-gcp-bucket", "", "GCP bucket name where transaction traces are stored")
 	flag.Parse()
 
 	if coinbase == "" {
@@ -212,6 +217,8 @@ func FromFlags() (*Config, error) {
 		cfg.ForceStartCadenceHeight = forceStartHeight
 	}
 
+	cfg.TracesEnabled = cfg.TracesBucketName != ""
+
 	// todo validate Config values
 	return cfg, nil
 }

go.mod

@@ -3,6 +3,7 @@ module github.com/onflow/flow-evm-gateway
 go 1.22
 
 require (
+	cloud.google.com/go/storage v1.36.0
 	github.com/cockroachdb/pebble v0.0.0-20230928194634-aa077af62593
 	github.com/goccy/go-json v0.10.2
 	github.com/hashicorp/golang-lru/v2 v2.0.7
@@ -13,12 +14,17 @@ require (
 	github.com/rs/cors v1.8.0
 	github.com/rs/zerolog v1.31.0
 	github.com/sethvargo/go-limiter v1.0.0
+	github.com/sethvargo/go-retry v0.2.3
 	github.com/stretchr/testify v1.9.0
 	golang.org/x/exp v0.0.0-20240119083558-1b970713d09a
 	golang.org/x/sync v0.6.0
 )
 
 require (
+	cloud.google.com/go v0.112.0 // indirect
+	cloud.google.com/go/compute v1.24.0 // indirect
+	cloud.google.com/go/compute/metadata v0.2.3 // indirect
+	cloud.google.com/go/iam v1.1.6 // indirect
 	github.com/DataDog/zstd v1.5.2 // indirect
 	github.com/Microsoft/go-winio v0.6.1 // indirect
 	github.com/SaveTheRbtz/mph v0.1.1-0.20240117162131-4166ec7869bc // indirect
@@ -48,6 +54,7 @@ require (
 	github.com/ef-ds/deque v1.0.4 // indirect
 	github.com/ethereum/c-kzg-4844 v0.4.0 // indirect
 	github.com/ethereum/go-ethereum v1.13.10 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.4.1-0.20230228173756-c0c9f774e40c // indirect
 	github.com/fxamacker/circlehash v0.3.0 // indirect
@@ -60,9 +67,13 @@ require (
 	github.com/gofrs/flock v0.8.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/glog v1.2.0 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
+	github.com/google/s2a-go v0.1.7 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
+	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.0 // indirect
@@ -138,7 +149,6 @@ require (
 	github.com/psiemens/sconfig v0.1.0 // indirect
 	github.com/rivo/uniseg v0.4.4 // indirect
 	github.com/rogpeppe/go-internal v1.10.0 // indirect
-	github.com/sethvargo/go-retry v0.2.3 // indirect
 	github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible // indirect
 	github.com/slok/go-http-metrics v0.10.0 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
@@ -163,6 +173,9 @@ require (
 	github.com/vmihailenco/tagparser v0.1.1 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/zeebo/blake3 v0.2.3 // indirect
+	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.47.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.47.0 // indirect
 	go.opentelemetry.io/otel v1.24.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.21.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.21.0 // indirect
@@ -176,13 +189,16 @@ require (
 	golang.org/x/crypto v0.19.0 // indirect
 	golang.org/x/mod v0.14.0 // indirect
 	golang.org/x/net v0.21.0 // indirect
+	golang.org/x/oauth2 v0.17.0 // indirect
 	golang.org/x/sys v0.17.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	golang.org/x/tools v0.17.0 // indirect
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
 	gonum.org/v1/gonum v0.14.0 // indirect
+	google.golang.org/api v0.162.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
+	google.golang.org/genproto v0.0.0-20240227224415-6ceb2ff114de // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240227224415-6ceb2ff114de // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240227224415-6ceb2ff114de // indirect
 	google.golang.org/grpc v1.63.2 // indirect

go.sum

@@ -1116,6 +1116,8 @@ github.com/cncf/xds/go v0.0.0-20230105202645-06c439db220b/go.mod h1:eXthEFrGJvWH
 github.com/cncf/xds/go v0.0.0-20230310173818-32f1caf87195/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20230428030218-4003588d1b74/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20230607035331-e9ce68804cb4/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa h1:jQCWAUqqlij9Pgj2i/PB79y4KOPYVyFYdROxgaCwdTQ=
+github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa/go.mod h1:x/1Gn8zydmfq8dk6e9PdstVsDgu9RuyIIJqAaF//0IM=
 github.com/cockroachdb/datadriven v1.0.0/go.mod h1:5Ib8Meh+jk1RlHIXej6Pzevx/NLlNvQB9pmSBZErGA4=
 github.com/cockroachdb/datadriven v1.0.2/go.mod h1:a9RdTaap04u637JoCzcUoIcDmvwSUtcUFtT/C3kJlTU=
 github.com/cockroachdb/datadriven v1.0.3-0.20230413201302-be42291fc80f h1:otljaYPt5hWxV3MUfO5dFPFiOXg9CyG5/kCfayTqsJ4=
@@ -1250,6 +1252,8 @@ github.com/envoyproxy/protoc-gen-validate v0.10.0/go.mod h1:DRjgyB0I43LtJapqN6Ni
 github.com/envoyproxy/protoc-gen-validate v0.10.1/go.mod h1:DRjgyB0I43LtJapqN6NiRwroiAU2PaFuvk/vjgh61ss=
 github.com/envoyproxy/protoc-gen-validate v1.0.1/go.mod h1:0vj8bNkYbSTNS2PIyH87KZaeN4x9zpL9Qt8fQC7d+vs=
 github.com/envoyproxy/protoc-gen-validate v1.0.2/go.mod h1:GpiZQP3dDbg4JouG/NNS7QWXpgx6x8QiMKdmN72jogE=
+github.com/envoyproxy/protoc-gen-validate v1.0.4 h1:gVPz/FMfvh57HdSJQyvBtF00j8JU4zdyUgIUNhlgg0A=
+github.com/envoyproxy/protoc-gen-validate v1.0.4/go.mod h1:qys6tmnRsYrQqIhm2bvKZH4Blx/1gTIZ2UKVY1M+Yew=
 github.com/etcd-io/bbolt v1.3.3/go.mod h1:ZF2nL25h33cCyBtcyWeZ2/I3HQOfTP+0PIEvHjkjCrw=
 github.com/ethereum/c-kzg-4844 v0.4.0 h1:3MS1s4JtA868KpJxroZoepdV0ZKBp3u/O5HcZ7R3nlY=
 github.com/ethereum/c-kzg-4844 v0.4.0/go.mod h1:VewdlzQmpT5QSrVhbBuGoCdFJkpaJlO1aQputP83wc0=
@@ -1485,10 +1489,12 @@ github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
 github.com/google/gofuzz v1.1.1-0.20200604201612-c04b05f3adfa/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
 github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
 github.com/google/martian/v3 v3.1.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
 github.com/google/martian/v3 v3.2.1/go.mod h1:oBOf6HBosgwRXnUGWUB05QECsc6uvmMiJ3+6W4l/CUk=
+github.com/google/martian/v3 v3.3.2 h1:IqNFLAmvJOgVlpdEBiQbDc2EwKW77amAycfTuWKdfvw=
 github.com/google/martian/v3 v3.3.2/go.mod h1:oBOf6HBosgwRXnUGWUB05QECsc6uvmMiJ3+6W4l/CUk=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@@ -1534,7 +1540,6 @@ github.com/googleapis/enterprise-certificate-proxy v0.2.4/go.mod h1:AwSRAtLfXpU5
 github.com/googleapis/enterprise-certificate-proxy v0.2.5/go.mod h1:RxW0N9901Cko1VOCW3SXCpWP+mlIEkk2tP7jnHy9a3w=
 github.com/googleapis/enterprise-certificate-proxy v0.3.2 h1:Vie5ybvEvT75RniqhfFxPRy3Bf7vr3h0cechB90XaQs=
 github.com/googleapis/enterprise-certificate-proxy v0.3.2/go.mod h1:VLSiSSBs/ksPL8kq3OBOQ6WRI2QnaFynd1DCjZ62+V0=
-github.com/googleapis/gax-go v0.0.0-20161107002406-da06d194a00e h1:CYRpN206UTHUinz3VJoLaBdy1gEGeJNsqT0mvswDcMw=
 github.com/googleapis/gax-go v0.0.0-20161107002406-da06d194a00e/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=

models/events.go

@@ -1,12 +1,13 @@
 package models
 
 import (
+	"strings"
+
 	"github.com/onflow/cadence"
 	"github.com/onflow/flow-go-sdk"
 	"github.com/onflow/flow-go/fvm/evm/types"
 	gethTypes "github.com/onflow/go-ethereum/core/types"
 	"golang.org/x/exp/slices"
-	"strings"
 )
 
 // isBlockExecutedEvent checks whether the given event contains block executed data.
@@ -103,6 +104,11 @@ func (c *CadenceEvents) CadenceHeight() uint64 {
 	return c.events.Height
 }
 
+// CadenceBlockID returns the Flow Cadence block ID.
+func (c *CadenceEvents) CadenceBlockID() flow.Identifier {
+	return c.events.BlockID
+}
+
 // Length of the Cadence events emitted.
 func (c *CadenceEvents) Length() int {
 	return len(c.events.Events)