Multi-spork client access support

Makefile

@@ -22,7 +22,7 @@ generate:
 	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 --all --dir=services/events --output=services/events/mocks
+	mockery --all --dir=services/ingestion --output=services/ingestion/mocks
 	mockery --dir=models --name=Engine --output=models/mocks
 
 .PHONY: ci

README.md

@@ -90,23 +90,26 @@ it should return:
 
 The application can be configured using the following flags at runtime:
 
-| Flag                      | Default Value    | Description                                                                                                            |
-|---------------------------|------------------|------------------------------------------------------------------------------------------------------------------------|
-| `--database-dir`          | `./db`           | Path to the directory for the database.                                                                                |
-| `--rpc-host`              | `localhost`      | Host for the JSON RPC API server.                                                                                      |
-| `--rpc-port`              | `8545`           | Port for the JSON RPC API server.                                                                                      |
-| `--access-node-grpc-host` | `localhost:3569` | Host to the Flow access node (AN) gRPC API.                                                                            |
-| `--evm-network-id`        | `testnet`        | EVM network ID (options: `testnet`, `mainnet`).                                                                        |
-| `--flow-network-id`       | `emulator`       | Flow network ID (options: `emulator`, `previewnet`).                                                                   |
-| `--coinbase`              | (required)       | Coinbase address to use for fee collection.                                                                            |
-| `--gas-price`             | `1`              | Static gas price used for EVM transactions.                                                                            |
-| `--coa-address`           | (required)       | Flow address that holds COA account used for submitting transactions.                                                  |
-| `--coa-key`               | (required)       | *WARNING*: Do not use this flag in production! Private key value for the COA address used for submitting transactions. |
-| `--coa-resource-create`   | `false`          | Auto-create the COA resource in the Flow COA account provided if one doesn't exist.                                    |
-| `--log-level`             | `debug`          | Define verbosity of the log output ('debug', 'info', 'error')                                                          |
-| `--stream-limit`          | 10               | Rate-limits the events sent to the client within one second                                                            |
-| `--stream-timeout`        | 3sec             | Defines the timeout in seconds the server waits for the event to be sent to the client                                 |
-
+| Flag                        | Default Value    | Description                                                                                                                                     |
+|-----------------------------|------------------|-------------------------------------------------------------------------------------------------------------------------------------------------|
+| `--database-dir`            | `./db`           | Path to the directory for the database.                                                                                                         |
+| `--rpc-host`                | `localhost`      | Host for the JSON RPC API server.                                                                                                               |
+| `--rpc-port`                | `8545`           | Port for the JSON RPC API server.                                                                                                               |
+| `--access-node-grpc-host`   | `localhost:3569` | Host to the current spork Flow access node (AN) gRPC API.                                                                                       |
+| `--access-node-spork-hosts` |                  | Previous spork AN hosts, defined following the schema: `{latest height}@{host}` as comma separated list (e.g. `"200@host-1.com,300@host2.com"`) |
+| `--evm-network-id`          | `testnet`        | EVM network ID (options: `testnet`, `mainnet`).                                                                                                 |
+| `--flow-network-id`         | `emulator`       | Flow network ID (options: `emulator`, `previewnet`).                                                                                            |
+| `--coinbase`                | (required)       | Coinbase address to use for fee collection.                                                                                                     |
+| `--init-cadence-height`     | 0                | Define the Cadence block height at which to start the indexing.                                                                                 |
+| `--gas-price`               | `1`              | Static gas price used for EVM transactions.                                                                                                     |
+| `--coa-address`             | (required)       | Flow address that holds COA account used for submitting transactions.                                                                           |
+| `--coa-key`                 | (required)       | *WARNING*: Do not use this flag in production! Private key value for the COA address used for submitting transactions.                          |
+| `--coa-key-file`            |                  | File path that contains JSON array of COA keys used in key-rotation mechanism, this is exclusive with `coa-key` flag.                           |
+| `--coa-resource-create`     | `false`          | Auto-create the COA resource in the Flow COA account provided if one doesn't exist.                                                             |
+| `--log-level`               | `debug`          | Define verbosity of the log output ('debug', 'info', 'error')                                                                                   |
+| `--stream-limit`            | 10               | Rate-limits the events sent to the client within one second                                                                                     |
+| `--stream-timeout`          | 3sec             | Defines the timeout in seconds the server waits for the event to be sent to the client                                                          |
+| `--filter-expiry`           | `5m`             | Filter defines the time it takes for an idle filter to expire                                                                                   |
 
 ## Getting Started
 

bootstrap/bootstrap.go

@@ -5,6 +5,13 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/onflow/flow-go-sdk/access"
+	"github.com/onflow/flow-go-sdk/access/grpc"
+	"github.com/onflow/flow-go-sdk/crypto"
+	broadcast "github.com/onflow/flow-go/engine"
+	"github.com/onflow/go-ethereum/rpc"
+	"github.com/rs/zerolog"
+
 	"github.com/onflow/flow-evm-gateway/api"
 	"github.com/onflow/flow-evm-gateway/config"
 	"github.com/onflow/flow-evm-gateway/models"
@@ -13,11 +20,6 @@ import (
 	"github.com/onflow/flow-evm-gateway/storage"
 	storageErrs "github.com/onflow/flow-evm-gateway/storage/errors"
 	"github.com/onflow/flow-evm-gateway/storage/pebble"
-	"github.com/onflow/flow-go-sdk/access/grpc"
-	"github.com/onflow/flow-go-sdk/crypto"
-	broadcast "github.com/onflow/flow-go/engine"
-	"github.com/onflow/go-ethereum/rpc"
-	"github.com/rs/zerolog"
 )
 
 func Start(ctx context.Context, cfg *config.Config) error {
@@ -107,7 +109,23 @@ func startIngestion(
 ) error {
 	logger.Info().Msg("starting up event ingestion")
 
-	client, err := grpc.NewClient(cfg.AccessNodeGRPCHost)
+	currentSporkClient, err := grpc.NewClient(cfg.AccessNodeHost)
+	if err != nil {
+		return fmt.Errorf("failed to create client connection for host: %s, with error: %w", cfg.AccessNodeHost, err)
+	}
+
+	// if we provided access node previous spork hosts add them to the client
+	pastSporkClients := make([]access.Client, len(cfg.AccessNodePreviousSporkHosts))
+	for i, host := range cfg.AccessNodePreviousSporkHosts {
+		grpcClient, err := grpc.NewClient(host)
+		if err != nil {
+			return fmt.Errorf("failed to create client connection for host: %s, with error: %w", host, err)
+		}
+
+		pastSporkClients[i] = grpcClient
+	}
+
+	client, err := requester.NewCrossSporkClient(currentSporkClient, pastSporkClients, logger)
 	if err != nil {
 		return err
 	}
@@ -123,7 +141,7 @@ func startIngestion(
 	}
 
 	// make sure the provided block to start the indexing can be loaded
-	_, err = client.GetBlockByHeight(context.Background(), latestCadenceHeight)
+	_, err = client.GetBlockHeaderByHeight(context.Background(), latestCadenceHeight)
 	if err != nil {
 		return fmt.Errorf("failed to get provided cadence height: %w", err)
 	}
@@ -134,7 +152,7 @@ func startIngestion(
 		Uint64("missed-heights", blk.Height-latestCadenceHeight).
 		Msg("indexing cadence height information")
 
-	subscriber := ingestion.NewRPCSubscriber(client, cfg.FlowNetworkID)
+	subscriber := ingestion.NewRPCSubscriber(client, cfg.FlowNetworkID, logger)
 	engine := ingestion.NewEventIngestionEngine(
 		subscriber,
 		blocks,
@@ -180,7 +198,7 @@ func startServer(
 
 	srv := api.NewHTTPServer(l, rpc.DefaultHTTPTimeouts)
 
-	client, err := grpc.NewClient(cfg.AccessNodeGRPCHost)
+	client, err := grpc.NewClient(cfg.AccessNodeHost)
 	if err != nil {
 		return err
 	}

config/config.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"math/big"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/goccy/go-json"
@@ -29,8 +30,10 @@ const LiveNetworkInitCadenceHeght = uint64(1)
 type Config struct {
 	// DatabaseDir is where the database should be stored.
 	DatabaseDir string
-	// AccessNodeGRPCHost defines the Flow network AN host.
-	AccessNodeGRPCHost string
+	// AccessNodeHost defines the current spork Flow network AN host.
+	AccessNodeHost string
+	// AccessNodePreviousSporkHosts contains a list of the ANs hosts for each spork
+	AccessNodePreviousSporkHosts []string
 	// GRPCPort for the RPC API server
 	RPCPort int
 	// GRPCHost for the RPC API server
@@ -71,15 +74,16 @@ type Config struct {
 
 func FromFlags() (*Config, error) {
 	cfg := &Config{}
-	var evmNetwork, coinbase, gas, coa, key, keysPath, flowNetwork, logLevel, filterExpiry string
+	var evmNetwork, coinbase, gas, coa, key, keysPath, flowNetwork, logLevel, filterExpiry, accessSporkHosts string
 	var streamTimeout int
 	var initHeight, forceStartHeight uint64
 
 	// parse from flags
 	flag.StringVar(&cfg.DatabaseDir, "database-dir", "./db", "Path to the directory for the database")
 	flag.StringVar(&cfg.RPCHost, "rpc-host", "", "Host for the RPC API server")
 	flag.IntVar(&cfg.RPCPort, "rpc-port", 8545, "Port for the RPC API server")
-	flag.StringVar(&cfg.AccessNodeGRPCHost, "access-node-grpc-host", "localhost:3569", "Host to the flow access node gRPC API")
+	flag.StringVar(&cfg.AccessNodeHost, "access-node-grpc-host", "localhost:3569", "Host to the flow access node gRPC API")
+	flag.StringVar(&accessSporkHosts, "access-node-spork-hosts", "", `Previous spork AN hosts, defined following the schema: {host1},{host2} as a comma separated list (e.g. "host-1.com,host2.com")`)
 	flag.StringVar(&evmNetwork, "evm-network-id", "previewnet", "EVM network ID (previewnet, testnet, mainnet)")
 	flag.StringVar(&flowNetwork, "flow-network-id", "flow-emulator", "Flow network ID (flow-emulator, flow-previewnet)")
 	flag.StringVar(&coinbase, "coinbase", "", "Coinbase address to use for fee collection")
@@ -184,6 +188,11 @@ func FromFlags() (*Config, error) {
 	}
 	cfg.FilterExpiry = exp
 
+	if accessSporkHosts != "" {
+		heightHosts := strings.Split(accessSporkHosts, ",")
+		cfg.AccessNodePreviousSporkHosts = append(cfg.AccessNodePreviousSporkHosts, heightHosts...)
+	}
+
 	if forceStartHeight != 0 {
 		cfg.ForceStartCadenceHeight = forceStartHeight
 	}

models/events.go

@@ -107,3 +107,21 @@ func (c *CadenceEvents) CadenceHeight() uint64 {
 func (c *CadenceEvents) Length() int {
 	return len(c.events.Events)
 }
+
+// BlockEvents is a wrapper around events streamed, and it also contains an error
+type BlockEvents struct {
+	Events *CadenceEvents
+	Err    error
+}
+
+func NewBlockEvents(events flow.BlockEvents) BlockEvents {
+	return BlockEvents{
+		Events: NewCadenceEvents(events),
+	}
+}
+
+func NewBlockEventsError(err error) BlockEvents {
+	return BlockEvents{
+		Err: err,
+	}
+}

services/ingestion/engine.go

@@ -2,12 +2,10 @@ package ingestion
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	"github.com/onflow/flow-evm-gateway/models"
 	"github.com/onflow/flow-evm-gateway/storage"
-	"github.com/onflow/flow-go-sdk"
 	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/fvm/evm/types"
 	gethTypes "github.com/onflow/go-ethereum/core/types"
@@ -96,45 +94,21 @@ func (e *Engine) Run(ctx context.Context) error {
 
 	e.log.Info().Uint64("start-cadence-height", latestCadence).Msg("starting ingestion")
 
-	events, errs, err := e.subscriber.Subscribe(ctx, latestCadence)
-	if err != nil {
-		return fmt.Errorf("failed to subscribe to events: %w", err)
-	}
-
 	e.status.MarkReady()
 
-	for {
-		select {
-		case <-ctx.Done():
-			e.log.Info().Msg("event ingestion received done signal")
-			return nil
-
-		case blockEvents, ok := <-events:
-			if !ok {
-				if ctx.Err() != nil {
-					return ctx.Err()
-				}
-				return models.ErrDisconnected
-			}
-
-			err = e.processEvents(blockEvents)
-			if err != nil {
-				e.log.Error().Err(err).Msg("failed to process EVM events")
-				return err
-			}
-
-		case err, ok := <-errs:
-			if !ok {
-				if ctx.Err() != nil {
-					return ctx.Err()
-				}
-
-				return models.ErrDisconnected
-			}
-
-			return errors.Join(err, models.ErrDisconnected)
+	for events := range e.subscriber.Subscribe(ctx, latestCadence) {
+		if events.Err != nil {
+			return fmt.Errorf("failure in event subscription: %w", events.Err)
+		}
+
+		err = e.processEvents(events.Events)
+		if err != nil {
+			e.log.Error().Err(err).Msg("failed to process EVM events")
+			return err
 		}
 	}
+
+	return nil
 }
 
 // processEvents converts the events to block and transactions and indexes them.
@@ -149,14 +123,12 @@ func (e *Engine) Run(ctx context.Context) error {
 // https://github.com/onflow/flow-go/blob/master/fvm/evm/types/events.go
 //
 // Any error is unexpected and fatal.
-func (e *Engine) processEvents(blockEvents flow.BlockEvents) error {
+func (e *Engine) processEvents(events *models.CadenceEvents) error {
 	e.log.Debug().
-		Uint64("cadence-height", blockEvents.Height).
-		Int("cadence-event-length", len(blockEvents.Events)).
+		Uint64("cadence-height", events.CadenceHeight()).
+		Int("cadence-event-length", events.Length()).
 		Msg("received new cadence evm events")
 
-	events := models.NewCadenceEvents(blockEvents)
-
 	// if heartbeat interval with no data still update the cadence height
 	if events.Empty() {
 		if err := e.blocks.SetLatestCadenceHeight(events.CadenceHeight()); err != nil {