Merge branch 'main' into eric/reduce-logging

.github/workflows/build.yml

@@ -14,7 +14,7 @@ jobs:
   build:
     strategy:
       matrix:
-        image: [{file: "connect.e2e.Dockerfile", name: "connect-simapp"}, {file: "slinky.sidecar.prod.Dockerfile", name: "slinky-sidecar"}, {file: "connect.sidecar.prod.Dockerfile", name: "connect-sidecar"}, {file: "connect.local.Dockerfile", name: "connect-testapp"}, {file: "connect.sidecar.e2e.Dockerfile", name: "connect-e2e-sidecar"}]
+        image: [{file: "connect.e2e.Dockerfile", name: "connect-simapp"}, {file: "slinky.sidecar.prod.Dockerfile", name: "slinky-sidecar"}, {file: "connect.sidecar.prod.Dockerfile", name: "connect-sidecar"}, {file: "connect.local.Dockerfile", name: "connect-testapp"}]
     runs-on: ubuntu-latest-m
     permissions:
       contents: read
@@ -102,13 +102,13 @@ jobs:
       - name: Update the image tags
         uses: skip-mev/gitops-actions/update-values@main
         with:
-          service: "slinky"
+          service: "connect"
           app_id: ${{ vars.DEPLOYER_APP_ID }}
           app_private_key: ${{ secrets.DEPLOYER_PRIVATE_KEY }}
-          manifests_repo: "slinky-manifests"
+          manifests_repo: "slinky-sensitive-manifests"
           values_file_name: "values-dev.yaml"
           modified_values: |
             {
               ".chain.image.tag": "${{ fromJSON(steps.matrix_output.outputs.result).image['connect-testapp'] }}",
-              ".sidecar.image.tag": "${{ fromJSON(steps.matrix_output.outputs.result).image['connect-e2e-sidecar'] }}",
+              ".sidecar.image.tag": "${{ fromJSON(steps.matrix_output.outputs.result).image['connect-sidecar'] }}",
             }

Makefile

@@ -89,7 +89,7 @@ install: tidy
 docker-build:
 	@echo "Building E2E Docker image..."
 	@DOCKER_BUILDKIT=1 $(DOCKER) build -t skip-mev/connect-e2e -f contrib/images/connect.e2e.Dockerfile .
-	@DOCKER_BUILDKIT=1 $(DOCKER) build -t skip-mev/connect-e2e-oracle -f contrib/images/connect.sidecar.dev.Dockerfile .
+	@DOCKER_BUILDKIT=1 $(DOCKER) build -t skip-mev/connect-e2e-oracle -f contrib/images/connect.sidecar.prod.Dockerfile .
 
 .PHONY: docker-build
 

oracle/config/api.go

@@ -40,6 +40,11 @@ type APIConfig struct {
 
 	// Name is the name of the provider that corresponds to this config.
 	Name string `json:"name"`
+
+	// MaxBlockHeightAge is the oldest an update from an on-chain data source can be without having its
+	// block height incremented.  In the case where a data source has exceeded this limit and the block
+	// height is not increasing, price reporting will be skipped until the block height increases.
+	MaxBlockHeightAge time.Duration `json:"maxBlockHeightAge"`
 }
 
 // Endpoint holds all data necessary for an API provider to connect to a given endpoint
@@ -123,5 +128,9 @@ func (c *APIConfig) ValidateBasic() error {
 		}
 	}
 
+	if c.MaxBlockHeightAge < 0 {
+		return fmt.Errorf("max_block_height_age cannot be negative")
+	}
+
 	return nil
 }

oracle/config/api_test.go

@@ -135,6 +135,36 @@ func TestAPIConfig(t *testing.T) {
 			},
 			expectedErr: false,
 		},
+		{
+			name: "good config with max_block_height_age",
+			config: config.APIConfig{
+				Enabled:           true,
+				Timeout:           time.Second,
+				Interval:          time.Second,
+				ReconnectTimeout:  time.Second,
+				MaxQueries:        1,
+				Name:              "test",
+				Endpoints:         []config.Endpoint{{URL: "http://test.com"}},
+				BatchSize:         1,
+				MaxBlockHeightAge: 10 * time.Second,
+			},
+			expectedErr: false,
+		},
+		{
+			name: "bad config with negative max_block_height_age",
+			config: config.APIConfig{
+				Enabled:           true,
+				Timeout:           time.Second,
+				Interval:          time.Second,
+				ReconnectTimeout:  time.Second,
+				MaxQueries:        1,
+				Name:              "test",
+				Endpoints:         []config.Endpoint{{URL: "http://test.com"}},
+				BatchSize:         1,
+				MaxBlockHeightAge: -10 * time.Second,
+			},
+			expectedErr: true,
+		},
 		{
 			name: "bad config with invalid endpoint (no url)",
 			config: config.APIConfig{

providers/apis/defi/ethmulticlient/multi_client.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/skip-mev/connect/v2/providers/apis/defi/types"
+
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/rpc"
 	"go.uber.org/zap"
@@ -23,6 +25,8 @@ type MultiRPCClient struct {
 
 	// underlying clients
 	clients []EVMClient
+
+	blockAgeChecker types.BlockAgeChecker
 }
 
 // NewMultiRPCClient returns a new MultiRPCClient.
@@ -32,9 +36,10 @@ func NewMultiRPCClient(
 	clients []EVMClient,
 ) EVMClient {
 	return &MultiRPCClient{
-		logger:  logger,
-		clients: clients,
-		api:     api,
+		logger:          logger,
+		clients:         clients,
+		api:             api,
+		blockAgeChecker: types.NewBlockAgeChecker(api.MaxBlockHeightAge),
 	}
 }
 
@@ -81,12 +86,20 @@ func NewMultiRPCClientFromEndpoints(
 	}
 
 	return &MultiRPCClient{
-		logger:  logger.With(zap.String("multi_client", api.Name)),
-		api:     api,
-		clients: clients,
+		logger:          logger.With(zap.String("multi_client", api.Name)),
+		api:             api,
+		clients:         clients,
+		blockAgeChecker: types.NewBlockAgeChecker(api.MaxBlockHeightAge),
 	}, nil
 }
 
+// define a result struct that go routines will populate and append to a slice when they complete their request.
+type result struct {
+	height  uint64
+	results []rpc.BatchElem
+	err     error
+}
+
 // BatchCallContext injects a call to eth_blockNumber, and makes batch calls to the underlying EVMClients.
 // It returns the response that has the greatest height from the eth_blockNumber call. An error is returned
 // only when no client was able to successfully provide a height or errored when sending the BatchCall.
@@ -95,15 +108,9 @@ func (m *MultiRPCClient) BatchCallContext(ctx context.Context, batchElems []rpc.
 		m.logger.Debug("BatchCallContext called with 0 elems")
 		return nil
 	}
-	// define a result struct that go routines will populate and append to a slice when they complete their request.
-	type result struct {
-		height  uint64
-		results []rpc.BatchElem
-	}
+
 	results := make([]result, len(m.clients))
 
-	// error slice to capture errors go routines encounter.
-	errs := make([]error, len(m.clients))
 	wg := new(sync.WaitGroup)
 	// this is the index of where we will have an eth_blockNumber call.
 	blockNumReqIndex := len(batchElems)
@@ -124,7 +131,8 @@ func (m *MultiRPCClient) BatchCallContext(ctx context.Context, batchElems []rpc.
 			// if there was an error, or if the block_num request didn't have result / errored
 			// we log the error and append to error slice.
 			if err != nil || req[blockNumReqIndex].Result == "" || req[blockNumReqIndex].Error != nil {
-				errs[i] = fmt.Errorf("endpoint request failed: %w, %w", err, req[blockNumReqIndex].Error)
+				resultErr := fmt.Errorf("endpoint request failed: %w, %w", err, req[blockNumReqIndex].Error)
+				results[i] = result{0, nil, resultErr}
 				m.logger.Debug(
 					"endpoint request failed",
 					zap.Error(err),
@@ -138,7 +146,8 @@ func (m *MultiRPCClient) BatchCallContext(ctx context.Context, batchElems []rpc.
 			// try to get the block number.
 			r, ok := req[blockNumReqIndex].Result.(*string)
 			if !ok {
-				errs[i] = fmt.Errorf("result from eth_blockNumber was not a string")
+				resultErr := fmt.Errorf("result from eth_blockNumber was not a string")
+				results[i] = result{0, nil, resultErr}
 				m.logger.Debug(
 					"result from eth_blockNumber was not a string",
 					zap.String("url", url),
@@ -149,7 +158,8 @@ func (m *MultiRPCClient) BatchCallContext(ctx context.Context, batchElems []rpc.
 			// decode the new height
 			height, err := hexutil.DecodeUint64(*r)
 			if err != nil { // if we can't decode the height, log an error.
-				errs[i] = fmt.Errorf("could not decode hex eth height: %w", err)
+				resultErr := fmt.Errorf("could not decode hex eth height: %w", err)
+				results[i] = result{0, nil, resultErr}
 				m.logger.Debug(
 					"could not decode hex eth height",
 					zap.String("url", url),
@@ -163,17 +173,31 @@ func (m *MultiRPCClient) BatchCallContext(ctx context.Context, batchElems []rpc.
 				zap.String("url", url),
 			)
 			// append the results, minus the appended eth_blockNumber request.
-			results[i] = result{height, req[:blockNumReqIndex]}
+			results[i] = result{height, req[:blockNumReqIndex], nil}
 		}(clientIdx)
 	}
 	wg.Wait()
 
+	filtered, err := m.filterResponses(results)
+	if err != nil {
+		return fmt.Errorf("error filtering responses: %w", err)
+	}
+
+	// copy the results from the results that had the largest height.
+	copy(batchElems, filtered)
+	return nil
+}
+
+// filterAccountsResponses chooses the rpc response with the highest block number.
+func (m *MultiRPCClient) filterResponses(responses []result) ([]rpc.BatchElem, error) {
 	// see which of the results had the largest height, and store the index of that result.
 	var (
 		maxHeight      uint64
 		maxHeightIndex int
+		errs           = make([]error, len(responses))
 	)
-	for i, res := range results {
+	for i, res := range responses {
+		errs[i] = res.err
 		if res.height > maxHeight {
 			maxHeight = res.height
 			maxHeightIndex = i
@@ -183,12 +207,17 @@ func (m *MultiRPCClient) BatchCallContext(ctx context.Context, batchElems []rpc.
 	if maxHeight == 0 {
 		err := errors.Join(errs...)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		// this should never happen... but who knows. maybe something terrible happened.
-		return errors.New("no errors were encountered, however no go routine was able to report a height")
+		return nil, errors.New("no errors were encountered, however no go routine was able to report a height")
+
 	}
-	// copy the results from the results that had the largest height.
-	copy(batchElems, results[maxHeightIndex].results)
-	return nil
+
+	// check the block height
+	if valid := m.blockAgeChecker.IsHeightValid(maxHeight); !valid {
+		return nil, fmt.Errorf("height %d is stale and older than %d", maxHeight, m.api.MaxBlockHeightAge)
+	}
+
+	return responses[maxHeightIndex].results, nil
 }

providers/apis/defi/osmosis/client.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/skip-mev/connect/v2/oracle/config"
 	"github.com/skip-mev/connect/v2/pkg/http"
+	"github.com/skip-mev/connect/v2/providers/apis/defi/types"
 	"github.com/skip-mev/connect/v2/providers/base/api/metrics"
 )
 
@@ -124,6 +125,8 @@ type MultiClientImpl struct {
 	apiMetrics metrics.APIMetrics
 
 	clients []Client
+
+	blockAgeChecker types.BlockAgeChecker
 }
 
 // NewMultiClient creates a new Client.
@@ -150,10 +153,11 @@ func NewMultiClient(
 	}
 
 	return &MultiClientImpl{
-		logger:     logger,
-		api:        api,
-		apiMetrics: apiMetrics,
-		clients:    clients,
+		logger:          logger,
+		api:             api,
+		apiMetrics:      apiMetrics,
+		clients:         clients,
+		blockAgeChecker: types.NewBlockAgeChecker(api.MaxBlockHeightAge),
 	}, nil
 }
 
@@ -190,10 +194,11 @@ func NewMultiClientFromEndpoints(
 	}
 
 	return &MultiClientImpl{
-		logger:     logger,
-		api:        api,
-		apiMetrics: apiMetrics,
-		clients:    clients,
+		logger:          logger,
+		api:             api,
+		apiMetrics:      apiMetrics,
+		clients:         clients,
+		blockAgeChecker: types.NewBlockAgeChecker(api.MaxBlockHeightAge),
 	}, nil
 }
 
@@ -225,11 +230,11 @@ func (mc *MultiClientImpl) SpotPrice(ctx context.Context, poolID uint64, baseAss
 
 	wg.Wait()
 
-	return filterSpotPriceResponses(resps)
+	return mc.filterSpotPriceResponses(resps)
 }
 
 // filterSpotPriceResponses chooses the response with the highest block height.
-func filterSpotPriceResponses(responses []WrappedSpotPriceResponse) (WrappedSpotPriceResponse, error) {
+func (mc *MultiClientImpl) filterSpotPriceResponses(responses []WrappedSpotPriceResponse) (WrappedSpotPriceResponse, error) {
 	if len(responses) == 0 {
 		return WrappedSpotPriceResponse{}, fmt.Errorf("no responses found")
 	}
@@ -244,5 +249,10 @@ func filterSpotPriceResponses(responses []WrappedSpotPriceResponse) (WrappedSpot
 		}
 	}
 
+	// check the block height
+	if valid := mc.blockAgeChecker.IsHeightValid(highestHeight); !valid {
+		return WrappedSpotPriceResponse{}, fmt.Errorf("height %d is stale and older than %d", highestHeight, mc.api.MaxBlockHeightAge)
+	}
+
 	return responses[highestHeightIndex], nil
 }

providers/apis/defi/osmosis/types.go

@@ -127,6 +127,7 @@ var DefaultAPIConfig = config.APIConfig{
 			URL: "https://osmosis-api.polkachu.com",
 		},
 	},
+	MaxBlockHeightAge: 30 * time.Second,
 }
 
 type SpotPriceResponse struct {