support shard-based routing

.env

@@ -43,6 +43,7 @@ TEST_DATABASE_ENDPOINT_URL=localhost:5432
 TEST_PROXY_BACKEND_HOST_URL_MAP=localhost:7777>http://kava-validator:8545,localhost:7778>http://kava-pruning:8545
 TEST_PROXY_HEIGHT_BASED_ROUTING_ENABLED=true
 TEST_PROXY_PRUNING_BACKEND_HOST_URL_MAP=localhost:7777>http://kava-pruning:8545,localhost:7778>http://kava-pruning:8545
+TEST_PROXY_SHARD_BACKEND_HOST_URL_MAP=localhost:7777>10|http://kava-shard-10:8545|20|http://kava-shard-20:8545
 # What level of logging to use for service objects constructed during
 # unit tests
 TEST_SERVICE_LOG_LEVEL=ERROR
@@ -71,9 +72,12 @@ PROXY_BACKEND_HOST_URL_MAP=localhost:7777>http://kava-validator:8545,localhost:7
 # otherwise, it falls back to the value in PROXY_BACKEND_HOST_URL_MAP
 PROXY_HEIGHT_BASED_ROUTING_ENABLED=true
 PROXY_PRUNING_BACKEND_HOST_URL_MAP=localhost:7777>http://kava-pruning:8545,localhost:7778>http://kava-pruning:8545
+# enable shard routing for hosts defined in PROXY_SHARD_BACKEND_HOST_URL_MAP
+PROXY_SHARDED_ROUTING_ENABLED=true
+PROXY_SHARD_BACKEND_HOST_URL_MAP=localhost:7777>10|http://kava-shard-10:8545|20|http://kava-shard-20:8545
 # PROXY_MAXIMUM_REQ_BATCH_SIZE is a proxy-enforced limit on the number of subrequest in a batch
 PROXY_MAXIMUM_REQ_BATCH_SIZE=100
-# Configuration for the servcie to connect to it's database
+# Configuration for the service to connect to it's database
 DATABASE_NAME=postgres
 DATABASE_ENDPOINT_URL=postgres:5432
 DATABASE_USERNAME=postgres

architecture/PROXY_ROUTING.md

@@ -77,7 +77,7 @@ Now suppose you want multiple backends for the same host.
 
 The proxy service supports height-based routing to direct requests that only require the most recent
 block to a different cluster.
-This support is handled via the [`HeightShardingProxies` implementation](../service/shard.go#L16).
+This support is handled via the [`PruningOrDefaultProxies` implementation](../service/shard.go#L16).
 
 This is configured via the `PROXY_HEIGHT_BASED_ROUTING_ENABLED` and `PROXY_PRUNING_BACKEND_HOST_URL_MAP`
 environment variables.

ci.docker-compose.yml

@@ -23,11 +23,14 @@ services:
     env_file: .env
     environment:
       PROXY_HEIGHT_BASED_ROUTING_ENABLED: "true"
+      PROXY_SHARDED_ROUTING_ENABLED: "true"
       # use public testnet as backend origin server to avoid having
       # to self-host a beefy Github Action runner
       # to build and run a kava node each execution
       PROXY_BACKEND_HOST_URL_MAP: localhost:7777>https://evmrpcdata.internal.testnet.proxy.kava.io,localhost:7778>https://evmrpc.internal.testnet.proxy.kava.io
       PROXY_PRUNING_BACKEND_HOST_URL_MAP: localhost:7777>https://evmrpc.internal.testnet.proxy.kava.io
+      # fake the shards by defining shards with existing backends
+      PROXY_SHARD_BACKEND_HOST_URL_MAP: localhost:7777>10|https://evmrpc.internal.testnet.proxy.kava.io|20|https://evmrpc.internal.testnet.proxy.kava.io
       EVM_QUERY_SERVICE_URL: https://evmrpc.internal.testnet.proxy.kava.io
     ports:
       - "${PROXY_HOST_PORT}:${PROXY_CONTAINER_PORT}"

config/config.go

@@ -20,6 +20,9 @@ type Config struct {
 	EnableHeightBasedRouting                      bool
 	ProxyPruningBackendHostURLMapRaw              string
 	ProxyPruningBackendHostURLMap                 map[string]url.URL
+	EnableShardedRouting                          bool
+	ProxyShardBackendHostURLMapRaw                string
+	ProxyShardBackendHostURLMap                   map[string]IntervalURLMap
 	ProxyMaximumBatchSize                         int
 	EvmQueryServiceURL                            string
 	DatabaseName                                  string
@@ -65,6 +68,8 @@ const (
 	PROXY_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY         = "PROXY_BACKEND_HOST_URL_MAP"
 	PROXY_HEIGHT_BASED_ROUTING_ENABLED_KEY             = "PROXY_HEIGHT_BASED_ROUTING_ENABLED"
 	PROXY_PRUNING_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY = "PROXY_PRUNING_BACKEND_HOST_URL_MAP"
+	PROXY_SHARDED_ROUTING_ENABLED_ENVIRONMENT_KEY      = "PROXY_SHARDED_ROUTING_ENABLED"
+	PROXY_SHARD_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY   = "PROXY_SHARD_BACKEND_HOST_URL_MAP"
 	PROXY_MAXIMUM_BATCH_SIZE_ENVIRONMENT_KEY           = "PROXY_MAXIMUM_REQ_BATCH_SIZE"
 	DEFAULT_PROXY_MAXIMUM_BATCH_SIZE                   = 500
 	PROXY_SERVICE_PORT_ENVIRONMENT_KEY                 = "PROXY_SERVICE_PORT"
@@ -220,6 +225,50 @@ func ParseRawProxyBackendHostURLMap(raw string) (map[string]url.URL, error) {
 	return hostURLMap, combinedErr
 }
 
+// ParseRawShardRoutingBackendHostURLMap attempts to parse backend host URL mapping for shards.
+// The shard map is a map of host name => (map of end block => backend route)
+// returning the mapping and error (if any)
+func ParseRawShardRoutingBackendHostURLMap(raw string) (map[string]IntervalURLMap, error) {
+	parsed := make(map[string]IntervalURLMap)
+	hostConfigs := strings.Split(raw, ",")
+	for _, hc := range hostConfigs {
+		pieces := strings.Split(hc, ">")
+		if len(pieces) != 2 {
+			return parsed, fmt.Errorf("expected shard definition like <host>:<end-height>|<backend-route>, found '%s'", hc)
+		}
+
+		host := pieces[0]
+		endpointBackendValues := strings.Split(pieces[1], "|")
+		if len(endpointBackendValues)%2 != 0 {
+			return parsed, fmt.Errorf("unexpected <end-height>|<backend-route> sequence for %s: %s",
+				host, pieces[1],
+			)
+		}
+
+		backendByEndHeight := make(map[uint64]*url.URL, len(endpointBackendValues)/2)
+		for i := 0; i < len(endpointBackendValues); i += 2 {
+			endHeight, err := strconv.ParseUint(endpointBackendValues[i], 10, 64)
+			if err != nil || endHeight == 0 {
+				return parsed, fmt.Errorf("invalid shard end height (%s) for host %s: %s",
+					endpointBackendValues[i], host, err,
+				)
+			}
+
+			backendRoute, err := url.Parse(endpointBackendValues[i+1])
+			if err != nil || backendRoute.String() == "" {
+				return parsed, fmt.Errorf("invalid shard backend route (%s) for height %d of host %s: %s",
+					endpointBackendValues[i+1], endHeight, host, err,
+				)
+			}
+			backendByEndHeight[endHeight] = backendRoute
+		}
+
+		parsed[host] = NewIntervalURLMap(backendByEndHeight)
+	}
+
+	return parsed, nil
+}
+
 // ParseRawHostnameToHeaderValueMap attempts to parse mappings of hostname to corresponding header value.
 // For example hostname to access-control-allow-origin header value.
 func ParseRawHostnameToHeaderValueMap(raw string) (map[string]string, error) {
@@ -257,10 +306,12 @@ func ParseRawHostnameToHeaderValueMap(raw string) (map[string]string, error) {
 func ReadConfig() Config {
 	rawProxyBackendHostURLMap := os.Getenv(PROXY_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY)
 	rawProxyPruningBackendHostURLMap := os.Getenv(PROXY_PRUNING_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY)
+	rawProxyShardedBackendHostURLMap := os.Getenv(PROXY_SHARD_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY)
 	// best effort to parse, callers are responsible for validating
 	// before using any values read
 	parsedProxyBackendHostURLMap, _ := ParseRawProxyBackendHostURLMap(rawProxyBackendHostURLMap)
 	parsedProxyPruningBackendHostURLMap, _ := ParseRawProxyBackendHostURLMap(rawProxyPruningBackendHostURLMap)
+	parsedProxyShardedBackendHostURLMap, _ := ParseRawShardRoutingBackendHostURLMap(rawProxyShardedBackendHostURLMap)
 
 	whitelistedHeaders := os.Getenv(WHITELISTED_HEADERS_ENVIRONMENT_KEY)
 	parsedWhitelistedHeaders := strings.Split(whitelistedHeaders, ",")
@@ -282,6 +333,9 @@ func ReadConfig() Config {
 		EnableHeightBasedRouting:                      EnvOrDefaultBool(PROXY_HEIGHT_BASED_ROUTING_ENABLED_KEY, false),
 		ProxyPruningBackendHostURLMapRaw:              rawProxyPruningBackendHostURLMap,
 		ProxyPruningBackendHostURLMap:                 parsedProxyPruningBackendHostURLMap,
+		EnableShardedRouting:                          EnvOrDefaultBool(PROXY_HEIGHT_BASED_ROUTING_ENABLED_KEY, false),
+		ProxyShardBackendHostURLMapRaw:                rawProxyShardedBackendHostURLMap,
+		ProxyShardBackendHostURLMap:                   parsedProxyShardedBackendHostURLMap,
 		ProxyMaximumBatchSize:                         EnvOrDefaultInt(PROXY_MAXIMUM_BATCH_SIZE_ENVIRONMENT_KEY, DEFAULT_PROXY_MAXIMUM_BATCH_SIZE),
 		DatabaseName:                                  os.Getenv(DATABASE_NAME_ENVIRONMENT_KEY),
 		DatabaseEndpointURL:                           os.Getenv(DATABASE_ENDPOINT_URL_ENVIRONMENT_KEY),

config/config_test.go

@@ -1,11 +1,13 @@
 package config_test
 
 import (
+	"net/url"
 	"os"
 	"testing"
 
 	"github.com/kava-labs/kava-proxy-service/config"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -14,6 +16,8 @@ var (
 	proxyServiceBackendHostURLMap        = os.Getenv("TEST_PROXY_BACKEND_HOST_URL_MAP")
 	proxyServiceHeightBasedRouting       = os.Getenv("TEST_PROXY_HEIGHT_BASED_ROUTING_ENABLED")
 	proxyServicePruningBackendHostURLMap = os.Getenv("TEST_PROXY_PRUNING_BACKEND_HOST_URL_MAP")
+	proxyServiceShardedRoutingEnabled    = os.Getenv("TEST_PROXY_HEIGHT_BASED_ROUTING_ENABLED")
+	proxyServiceShardBackendHostURLMap   = os.Getenv("TEST_PROXY_SHARD_BACKEND_HOST_URL_MAP")
 )
 
 func TestUnitTestEnvODefaultReturnsDefaultIfEnvironmentVariableNotSet(t *testing.T) {
@@ -53,10 +57,36 @@ func TestUnitTestParseHostMapReturnsErrEmptyHostMapWhenEmpty(t *testing.T) {
 	assert.ErrorIs(t, err, config.ErrEmptyHostMap)
 }
 
+func TestUnitTestParseRawShardRoutingBackendHostURLMap(t *testing.T) {
+	parsed, err := config.ParseRawShardRoutingBackendHostURLMap("localhost:7777>10|http://kava-shard-10:8545|20|http://kava-shard-20:8545")
+	require.NoError(t, err)
+	expected := map[string]config.IntervalURLMap{
+		"localhost:7777": config.NewIntervalURLMap(map[uint64]*url.URL{
+			10: mustUrl("http://kava-shard-10:8545"),
+			20: mustUrl("http://kava-shard-20:8545"),
+		}),
+	}
+	require.Equal(t, expected, parsed)
+
+	_, err = config.ParseRawShardRoutingBackendHostURLMap("no-shard-def")
+	require.ErrorContains(t, err, "expected shard definition like <host>:<end-height>|<backend-route>")
+
+	_, err = config.ParseRawShardRoutingBackendHostURLMap("invalid-shard-def>odd|number|bad")
+	require.ErrorContains(t, err, "unexpected <end-height>|<backend-route> sequence for invalid-shard-def")
+
+	_, err = config.ParseRawShardRoutingBackendHostURLMap("invalid-height>NaN|backend-host")
+	require.ErrorContains(t, err, "invalid shard end height (NaN) for host invalid-height")
+
+	_, err = config.ParseRawShardRoutingBackendHostURLMap("invalid-backend-host>100|")
+	require.ErrorContains(t, err, "invalid shard backend route () for height 100 of host invalid-backend-host")
+}
+
 func setDefaultEnv() {
 	os.Setenv(config.PROXY_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY, proxyServiceBackendHostURLMap)
 	os.Setenv(config.PROXY_HEIGHT_BASED_ROUTING_ENABLED_KEY, proxyServiceHeightBasedRouting)
 	os.Setenv(config.PROXY_PRUNING_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY, proxyServicePruningBackendHostURLMap)
+	os.Setenv(config.PROXY_SHARDED_ROUTING_ENABLED_ENVIRONMENT_KEY, proxyServiceShardedRoutingEnabled)
+	os.Setenv(config.PROXY_SHARD_BACKEND_HOST_URL_MAP_ENVIRONMENT_KEY, proxyServiceShardBackendHostURLMap)
 	os.Setenv(config.PROXY_SERVICE_PORT_ENVIRONMENT_KEY, proxyServicePort)
 	os.Setenv(config.LOG_LEVEL_ENVIRONMENT_KEY, config.DEFAULT_LOG_LEVEL)
 }

config/intervalmap.go

@@ -0,0 +1,41 @@
+package config
+
+import (
+	"net/url"
+	"sort"
+)
+
+// IntervalURLMap stores URLs associated with a range of numbers.
+// The intervals are defined by their endpoints and must not overlap.
+// The intervals are exclusive of the endpoints.
+type IntervalURLMap struct {
+	UrlByEndHeight map[uint64]*url.URL
+	endpoints      []uint64
+}
+
+// NewIntervalURLMap creates a new IntervalMap from a map of interval endpoint => url.
+// The intervals are exclusive of their endpoint.
+// ie. if the lowest value endpoint in the map is 10, the interval is for all numbers 1 through 9.
+func NewIntervalURLMap(urlByEndHeight map[uint64]*url.URL) IntervalURLMap {
+	endpoints := make([]uint64, 0, len(urlByEndHeight))
+	for e := range urlByEndHeight {
+		endpoints = append(endpoints, e)
+	}
+	sort.Slice(endpoints, func(i, j int) bool { return endpoints[i] < endpoints[j] })
+
+	return IntervalURLMap{
+		UrlByEndHeight: urlByEndHeight,
+		endpoints:      endpoints,
+	}
+}
+
+// Lookup finds the value associated with the interval containing the number, if it exists.
+func (im *IntervalURLMap) Lookup(num uint64) (*url.URL, uint64, bool) {
+	i := sort.Search(len(im.endpoints), func(i int) bool { return im.endpoints[i] > num })
+
+	if i < len(im.endpoints) && num < im.endpoints[i] {
+		return im.UrlByEndHeight[im.endpoints[i]], im.endpoints[i], true
+	}
+
+	return nil, 0, false
+}

config/intervalmap_test.go

@@ -0,0 +1,56 @@
+package config_test
+
+import (
+	"fmt"
+	"net/url"
+	"testing"
+
+	"github.com/kava-labs/kava-proxy-service/config"
+	"github.com/stretchr/testify/require"
+)
+
+func mustUrl(s string) *url.URL {
+	u, err := url.Parse(s)
+	if err != nil {
+		panic(fmt.Sprintf("failed to parse url %s: %s", s, err))
+	}
+	return u
+}
+
+func TestUnitTestIntervalMap(t *testing.T) {
+	valueByEndpoint := map[uint64]*url.URL{
+		10:  mustUrl("A"),
+		20:  mustUrl("B"),
+		100: mustUrl("C"),
+	}
+	intervalmap := config.NewIntervalURLMap(valueByEndpoint)
+
+	testCases := []struct {
+		value           uint64
+		expectFound     bool
+		expectEndHeight uint64
+		expectResult    string
+	}{
+		{1, true, 10, "A"},
+		{9, true, 10, "A"},
+		{10, true, 20, "B"},
+		{15, true, 20, "B"},
+		{20, true, 100, "C"},
+		{75, true, 100, "C"},
+		{100, false, 0, ""},
+		{300, false, 0, ""},
+	}
+
+	for _, tc := range testCases {
+		t.Run(fmt.Sprintf("Lookup(%d)", tc.value), func(t *testing.T) {
+			result, endHeight, found := intervalmap.Lookup(tc.value)
+			require.Equal(t, tc.expectFound, found)
+			require.Equal(t, tc.expectEndHeight, endHeight)
+			if tc.expectResult == "" {
+				require.Nil(t, result)
+			} else {
+				require.Equal(t, tc.expectResult, result.String())
+			}
+		})
+	}
+}

docker-compose.yml

@@ -51,6 +51,34 @@ services:
       - "${KAVA_CONTAINER_COSMOS_RPC_PORT}"
       - "${KAVA_CONTAINER_EVM_RPC_PORT}"
 
+  # shards are in name only. they are standard peer nodes, but will only recieve traffic
+  # for a specific block range. kava-shard-10 receives requests for heights 1-9
+  kava-shard-10:
+    image: kava/kava:${KAVA_CONTAINER_TAG}
+    entrypoint: /docker/shared/kava-entrypoint.sh
+    env_file: .env
+    volumes:
+      - ./docker/shared:/docker/shared
+    # expose ports for other services to be able to connect to within
+    # the default docker-compose network
+    expose:
+      - "${KAVA_CONTAINER_COSMOS_RPC_PORT}"
+      - "${KAVA_CONTAINER_EVM_RPC_PORT}"
+
+  # shards are in name only. they are standard peer nodes, but will only recieve traffic
+  # for a specific block range. kava-shard-20 receives requests for heights 10-19
+  kava-shard-20:
+    image: kava/kava:${KAVA_CONTAINER_TAG}
+    entrypoint: /docker/shared/kava-entrypoint.sh
+    env_file: .env
+    volumes:
+      - ./docker/shared:/docker/shared
+    # expose ports for other services to be able to connect to within
+    # the default docker-compose network
+    expose:
+      - "${KAVA_CONTAINER_COSMOS_RPC_PORT}"
+      - "${KAVA_CONTAINER_EVM_RPC_PORT}"
+
   # run proxy service to observe, route, and scale requests to kava api endpoints
   proxy:
     build:

main_test.go

@@ -89,8 +89,9 @@ var (
 // search for any request metrics between startTime and time.Now() for particular request methods
 // if testedmethods is empty, all metrics in timeframe are returned.
 func findMetricsInWindowForMethods(db database.PostgresClient, startTime time.Time, testedmethods []string) []database.ProxiedRequestMetric {
+	extension := time.Duration(testExtendMetricWindowMs) * time.Millisecond
 	// add small buffer into future in case metrics are still being created
-	endTime := time.Now().Add(time.Duration(testExtendMetricWindowMs) * time.Millisecond)
+	endTime := time.Now().Add(extension)
 
 	var nextCursor int64
 	var proxiedRequestMetrics []database.ProxiedRequestMetric
@@ -130,6 +131,8 @@ func findMetricsInWindowForMethods(db database.PostgresClient, startTime time.Ti
 		}
 	}
 
+	// ensure next window has no overlap with current one
+	time.Sleep(extension)
 	return requestMetricsDuringRequestWindow
 }
 
@@ -422,14 +425,26 @@ func TestE2ETest_HeightBasedRouting(t *testing.T) {
 		{
 			name:        "request for non-latest height -> default",
 			method:      "eth_getBlockByNumber",
-			params:      []interface{}{"0x2", false},
+			params:      []interface{}{"0x15", false}, // block 21 is beyond shards
 			expectRoute: service.ResponseBackendDefault,
 		},
 		{
-			name:        "request for earliest height -> default",
+			name:        "request for height in 1st shard -> shard",
+			method:      "eth_getBlockByNumber",
+			params:      []interface{}{"0x2", false}, // block 2
+			expectRoute: service.ResponseBackendShard,
+		},
+		{
+			name:        "request for height in 2nd shard -> shard",
+			method:      "eth_getBlockByNumber",
+			params:      []interface{}{"0xF", false}, // block 15
+			expectRoute: service.ResponseBackendShard,
+		},
+		{
+			name:        "request for earliest height -> 1st shard",
 			method:      "eth_getBlockByNumber",
 			params:      []interface{}{"earliest", false},
-			expectRoute: service.ResponseBackendDefault,
+			expectRoute: service.ResponseBackendShard,
 		},
 		{
 			name:        "request for latest height -> pruning",