Multi-spork client access support

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 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                  |
+| `--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

@@ -99,11 +99,18 @@ func startIngestion(
 ) error {
 	logger.Info().Msg("starting up event ingestion")
 
-	client, err := grpc.NewClient(cfg.AccessNodeGRPCHost)
+	client, err := models.NewCrossSporkClient(cfg.AccessNodeHost, logger)
 	if err != nil {
 		return err
 	}
 
+	// if we provided access node previous spork hosts add them to the client
+	for height, host := range cfg.AccessNodePreviousSporkHosts {
+		if err := client.AddSpork(height, host); err != nil {
+			return fmt.Errorf("failed to add previous spork host to the client: %w", err)
+		}
+	}
+
 	blk, err := client.GetLatestBlock(context.Background(), false)
 	if err != nil {
 		return fmt.Errorf("failed to get latest cadence block: %w", err)
@@ -172,7 +179,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,8 @@ import (
 	"io"
 	"math/big"
 	"os"
+	"strconv"
+	"strings"
 	"time"
 
 	"github.com/goccy/go-json"
@@ -29,8 +31,11 @@ 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 map of latest heights for each spork,
+	// which can be accessed via the host of the AN provided
+	AccessNodePreviousSporkHosts map[uint64]string
 	// GRPCPort for the RPC API server
 	RPCPort int
 	// GRPCHost for the RPC API server
@@ -68,16 +73,19 @@ type Config struct {
 }
 
 func FromFlags() (*Config, error) {
-	cfg := &Config{}
-	var evmNetwork, coinbase, gas, coa, key, keysPath, flowNetwork, logLevel, filterExpiry string
+	cfg := &Config{
+		AccessNodePreviousSporkHosts: make(map[uint64]string),
+	}
+	var evmNetwork, coinbase, gas, coa, key, keysPath, flowNetwork, logLevel, filterExpiry, accessSporkHosts string
 	var streamTimeout int
 	var initHeight 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: {latest height}@{host} as comma separated list (e.g. "[email protected],[email protected]")`)
 	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")
@@ -181,6 +189,20 @@ func FromFlags() (*Config, error) {
 	}
 	cfg.FilterExpiry = exp
 
+	if accessSporkHosts != "" {
+		heightHosts := strings.Split(accessSporkHosts, ",")
+		for _, hh := range heightHosts {
+			v := strings.Split(hh, "@")
+			heightVal, host := v[0], v[1]
+			height, err := strconv.Atoi(heightVal)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse AN host height value for previous sporks, provided with --access-node-spork-hosts flag")
+			}
+
+			cfg.AccessNodePreviousSporkHosts[uint64(height)] = host
+		}
+	}
+
 	// todo validate Config values
 	return cfg, nil
 }

models/cross-spork_client.go

@@ -0,0 +1,124 @@
+package models
+
+import (
+	"context"
+	"fmt"
+	"github.com/onflow/cadence"
+	"github.com/onflow/flow-go-sdk"
+	"github.com/onflow/flow-go-sdk/access"
+	"github.com/onflow/flow-go-sdk/access/grpc"
+	"github.com/rs/zerolog"
+	"golang.org/x/exp/maps"
+	"golang.org/x/exp/slices"
+)
+
+// CrossSporkClient is a wrapper around the Flow AN client that can
+// access different AN APIs based on the height boundaries of the sporks.
+//
+// Each spork is defined with the last height included in that spork,
+// based on the list we know which AN client to use when requesting the data.
+//
+// Any API that supports cross-spork access must have a defined function
+// that shadows the original access Client function.
+type CrossSporkClient struct {
+	logger zerolog.Logger
+	// this map holds the last heights and clients for each spork
+	sporkHosts map[uint64]access.Client
+
+	access.Client
+}
+
+// NewCrossSporkClient creates a new instance of the client, it accepts the
+// host to the current spork AN API.
+func NewCrossSporkClient(currentSporkHost string, logger zerolog.Logger) (*CrossSporkClient, error) {
+	// add current spork AN host as the default client
+	client, err := grpc.NewClient(currentSporkHost)
+	if err != nil {
+		return nil, err
+	}
+
+	return &CrossSporkClient{
+		logger,
+		make(map[uint64]access.Client),
+		client,
+	}, nil
+}
+
+func (c *CrossSporkClient) AddSpork(lastHeight uint64, host string) error {
+	if _, ok := c.sporkHosts[lastHeight]; ok {
+		return fmt.Errorf("provided last height already exists")
+	}
+
+	client, err := grpc.NewClient(host)
+	if err != nil {
+		return err
+	}
+
+	c.sporkHosts[lastHeight] = client
+
+	c.logger.Info().
+		Uint64("spork-boundary", lastHeight).
+		Str("host", host).
+		Msg("added spork specific client")
+
+	return nil
+}
+
+// getClientForHeight returns the client for the given height. It starts by using the current spork client,
+// then iteratively checks the upper height boundaries in descending order and returns the last client
+// that still contains the given height within its upper height limit. If no client is found, it returns
+// the current spork client.
+// Please note that even if a client for provided height is found we don't guarantee the data being available
+// because it still might not have access to the height provided, because there might be other sporks with
+// lower height boundaries that we didn't configure for.
+// This would result in the error when using the client to access such data.
+func (c *CrossSporkClient) getClientForHeight(height uint64) access.Client {
+	heights := maps.Keys(c.sporkHosts)
+	slices.Sort(heights)    // order heights in ascending order
+	slices.Reverse(heights) // make it descending
+
+	// start by using the current spork client, then iterate all the upper height boundaries
+	// and find the last client that still contains the height in its upper height limit
+	client := c.Client
+	for _, upperBound := range heights {
+		if upperBound >= height {
+			client = c.sporkHosts[upperBound]
+
+			c.logger.Debug().
+				Uint64("spork-boundary", upperBound).
+				Msg("using previous spork client")
+		}
+	}
+
+	return client
+}
+
+func (c *CrossSporkClient) GetBlockByHeight(
+	ctx context.Context,
+	height uint64,
+) (*flow.Block, error) {
+	return c.
+		getClientForHeight(height).
+		GetBlockByHeight(ctx, height)
+}
+
+func (c *CrossSporkClient) ExecuteScriptAtBlockHeight(
+	ctx context.Context,
+	height uint64,
+	script []byte,
+	arguments []cadence.Value,
+) (cadence.Value, error) {
+	return c.
+		getClientForHeight(height).
+		ExecuteScriptAtBlockHeight(ctx, height, script, arguments)
+}
+
+func (c *CrossSporkClient) SubscribeEventsByBlockHeight(
+	ctx context.Context,
+	startHeight uint64,
+	filter flow.EventFilter,
+) (<-chan flow.BlockEvents, <-chan error, error) {
+	return c.
+		getClientForHeight(startHeight).
+		SubscribeEventsByBlockHeight(ctx, startHeight, filter)
+}

models/cross-spork_client_test.go

@@ -0,0 +1,53 @@
+package models
+
+import (
+	"context"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestCrossSporkClient_MultiClient(t *testing.T) {
+	clientHosts := []string{"test1.com", "test2.com", "test3.com"}
+
+	client, err := NewCrossSporkClient(clientHosts[0], zerolog.Nop())
+	require.NoError(t, err)
+
+	err = client.AddSpork(100, clientHosts[1])
+	require.NoError(t, err)
+
+	err = client.AddSpork(200, clientHosts[2])
+	require.NoError(t, err)
+
+	c := client.getClientForHeight(300)
+	require.NotNil(t, c)
+
+	ctx := context.Background()
+
+	// this height should use current spork client
+	_, err = client.GetBlockByHeight(ctx, 300)
+	require.ErrorContains(t, err, clientHosts[0])
+
+	// this height should use test2 client
+	_, err = client.GetBlockByHeight(ctx, 150)
+	require.ErrorContains(t, err, clientHosts[2])
+
+	// this height should use test3 client
+	_, err = client.GetBlockByHeight(ctx, 50)
+	require.ErrorContains(t, err, clientHosts[1])
+
+	// test boundaries are inclusive
+	_, err = client.GetBlockByHeight(ctx, 200)
+	require.ErrorContains(t, err, clientHosts[2])
+}
+
+func TestCrossSporkClient_ExistingHeight(t *testing.T) {
+	client, err := NewCrossSporkClient("host1.com", zerolog.Nop())
+	require.NoError(t, err)
+
+	err = client.AddSpork(100, "host2.com")
+	require.NoError(t, err)
+
+	err = client.AddSpork(100, "host3.com")
+	require.EqualError(t, err, "provided last height already exists")
+}

tests/helpers.go

@@ -129,21 +129,21 @@ func servicesSetup(t *testing.T) (emulator.Emulator, func()) {
 
 	// default config
 	cfg := &config.Config{
-		DatabaseDir:        t.TempDir(),
-		AccessNodeGRPCHost: "localhost:3569", // emulator
-		RPCPort:            8545,
-		RPCHost:            "127.0.0.1",
-		FlowNetworkID:      "flow-emulator",
-		EVMNetworkID:       evmTypes.FlowEVMPreviewNetChainID,
-		Coinbase:           common.HexToAddress(eoaTestAddress),
-		COAAddress:         service.Address,
-		COAKey:             service.PrivateKey,
-		CreateCOAResource:  false,
-		GasPrice:           new(big.Int).SetUint64(0),
-		LogLevel:           zerolog.DebugLevel,
-		LogWriter:          zerolog.NewConsoleWriter(),
-		StreamTimeout:      time.Second * 30,
-		StreamLimit:        10,
+		DatabaseDir:       t.TempDir(),
+		AccessNodeHost:    "localhost:3569", // emulator
+		RPCPort:           8545,
+		RPCHost:           "127.0.0.1",
+		FlowNetworkID:     "flow-emulator",
+		EVMNetworkID:      evmTypes.FlowEVMPreviewNetChainID,
+		Coinbase:          common.HexToAddress(eoaTestAddress),
+		COAAddress:        service.Address,
+		COAKey:            service.PrivateKey,
+		CreateCOAResource: false,
+		GasPrice:          new(big.Int).SetUint64(0),
+		LogLevel:          zerolog.DebugLevel,
+		LogWriter:         zerolog.NewConsoleWriter(),
+		StreamTimeout:     time.Second * 30,
+		StreamLimit:       10,
 	}
 
 	if !logOutput {

tests/integration_test.go

@@ -55,19 +55,19 @@ func Test_ConcurrentTransactionSubmission(t *testing.T) {
 	require.NoError(t, err)
 
 	cfg := &config.Config{
-		DatabaseDir:        t.TempDir(),
-		AccessNodeGRPCHost: grpcHost,
-		RPCPort:            8545,
-		RPCHost:            "127.0.0.1",
-		FlowNetworkID:      "flow-emulator",
-		EVMNetworkID:       types.FlowEVMTestNetChainID,
-		Coinbase:           eoaTestAccount,
-		COAAddress:         *createdAddr,
-		COAKeys:            keys,
-		CreateCOAResource:  true,
-		GasPrice:           new(big.Int).SetUint64(0),
-		LogLevel:           zerolog.DebugLevel,
-		LogWriter:          os.Stdout,
+		DatabaseDir:       t.TempDir(),
+		AccessNodeHost:    grpcHost,
+		RPCPort:           8545,
+		RPCHost:           "127.0.0.1",
+		FlowNetworkID:     "flow-emulator",
+		EVMNetworkID:      types.FlowEVMTestNetChainID,
+		Coinbase:          eoaTestAccount,
+		COAAddress:        *createdAddr,
+		COAKeys:           keys,
+		CreateCOAResource: true,
+		GasPrice:          new(big.Int).SetUint64(0),
+		LogLevel:          zerolog.DebugLevel,
+		LogWriter:         os.Stdout,
 	}
 
 	// todo change this test to use ingestion and emulator directly so we can completely remove