chore: backport https://github.com/skip-mev/connect/pull/705

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 @@
 	"fmt"
 	"sync"
 
+	"github.com/skip-mev/slinky/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 @@
 
 	// underlying clients
 	clients []EVMClient
+
+	blockAgeChecker types.BlockAgeChecker
 }
 
 // NewMultiRPCClient returns a new MultiRPCClient.
@@ -32,9 +36,10 @@
 	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 @@
 	}
 
 	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 @@
 		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 @@
 			// 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 @@
 			// 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 @@
 			// 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 @@
 				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 @@
 	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

@@ -4,17 +4,22 @@
 	"context"
 	"encoding/json"
 	"fmt"
-	"math/rand"
+	"strconv"
 	"sync"
 	"time"
 
 	"go.uber.org/zap"
 
 	"github.com/skip-mev/slinky/oracle/config"
 	"github.com/skip-mev/slinky/pkg/http"
+	"github.com/skip-mev/slinky/providers/apis/defi/types"
 	"github.com/skip-mev/slinky/providers/base/api/metrics"
 )
 
+const (
+	headerBlockHeight = "grpc-metadata-x-cosmos-block-height"
+)
+
 var (
 	_ Client = &ClientImpl{}
 	_ Client = &MultiClientImpl{}
@@ -28,7 +33,7 @@
 		poolID uint64,
 		baseAsset,
 		quoteAsset string,
-	) (SpotPriceResponse, error)
+	) (WrappedSpotPriceResponse, error)
 }
 
 // ClientImpl is an implementation of a client to Osmosis using a
@@ -74,31 +79,43 @@
 }
 
 // SpotPrice uses the underlying x/poolmanager client to access spot prices.
-func (c *ClientImpl) SpotPrice(ctx context.Context, poolID uint64, baseAsset, quoteAsset string) (SpotPriceResponse, error) {
+func (c *ClientImpl) SpotPrice(ctx context.Context, poolID uint64, baseAsset, quoteAsset string) (WrappedSpotPriceResponse, error) {
 	start := time.Now()
 	defer func() {
 		c.apiMetrics.ObserveProviderResponseLatency(c.api.Name, c.redactedURL, time.Since(start))
 	}()
 
 	url, err := CreateURL(c.endpoint.URL, poolID, baseAsset, quoteAsset)
 	if err != nil {
-		return SpotPriceResponse{}, err
+		return WrappedSpotPriceResponse{}, err
 	}
 
 	resp, err := c.httpClient.GetWithContext(ctx, url)
 	if err != nil {
-		return SpotPriceResponse{}, err
+		return WrappedSpotPriceResponse{}, err
 	}
 
 	c.apiMetrics.AddHTTPStatusCode(c.api.Name, resp)
 
+	var blockHeight uint64
+	heightStr := resp.Header.Get(headerBlockHeight)
+	if heightStr != "" {
+		blockHeight, err = strconv.ParseUint(heightStr, 10, 64)
+		if err != nil {
+			return WrappedSpotPriceResponse{}, fmt.Errorf("failed to parse block height: %w", err)
+		}
+	}
+
 	var spotPriceResponse SpotPriceResponse
 	if err := json.NewDecoder(resp.Body).Decode(&spotPriceResponse); err != nil {
-		return SpotPriceResponse{}, err
+		return WrappedSpotPriceResponse{}, err
 	}
 
 	c.apiMetrics.AddHTTPStatusCode(c.api.Name, resp)
-	return spotPriceResponse, nil
+	return WrappedSpotPriceResponse{
+		SpotPriceResponse: spotPriceResponse,
+		BlockHeight:       blockHeight,
+	}, nil
 }
 
 // MultiClientImpl is an Osmosis client that wraps a set of multiple Clients.
@@ -108,6 +125,8 @@
 	apiMetrics metrics.APIMetrics
 
 	clients []Client
+
+	blockAgeChecker types.BlockAgeChecker
 }
 
 // NewMultiClient creates a new Client.
@@ -134,10 +153,11 @@
 	}
 
 	return &MultiClientImpl{
-		logger:     logger,
-		api:        api,
-		apiMetrics: apiMetrics,
-		clients:    clients,
+		logger:          logger,
+		api:             api,
+		apiMetrics:      apiMetrics,
+		clients:         clients,
+		blockAgeChecker: types.NewBlockAgeChecker(api.MaxBlockHeightAge),
 	}, nil
 }
 
@@ -174,17 +194,18 @@
 	}
 
 	return &MultiClientImpl{
-		logger:     logger,
-		api:        api,
-		apiMetrics: apiMetrics,
-		clients:    clients,
+		logger:          logger,
+		api:             api,
+		apiMetrics:      apiMetrics,
+		clients:         clients,
+		blockAgeChecker: types.NewBlockAgeChecker(api.MaxBlockHeightAge),
 	}, nil
 }
 
 // SpotPrice delegates the request to all underlying clients and applies a filter to the
 // set of responses.
-func (mc *MultiClientImpl) SpotPrice(ctx context.Context, poolID uint64, baseAsset, quoteAsset string) (SpotPriceResponse, error) {
-	resps := make([]SpotPriceResponse, len(mc.clients))
+func (mc *MultiClientImpl) SpotPrice(ctx context.Context, poolID uint64, baseAsset, quoteAsset string) (WrappedSpotPriceResponse, error) {
+	resps := make([]WrappedSpotPriceResponse, len(mc.clients))
 
 	var wg sync.WaitGroup
 	wg.Add(len(mc.clients))
@@ -209,22 +230,29 @@
 
 	wg.Wait()
 
-	return filterSpotPriceResponses(resps)
+	return mc.filterSpotPriceResponses(resps)
 }
 
-// filterSpotPriceResponses currently just chooses a random response as there is no way to differentiate.
-func filterSpotPriceResponses(responses []SpotPriceResponse) (SpotPriceResponse, error) {
+// filterSpotPriceResponses chooses the response with the highest block height.
+func (mc *MultiClientImpl) filterSpotPriceResponses(responses []WrappedSpotPriceResponse) (WrappedSpotPriceResponse, error) {
 	if len(responses) == 0 {
-		return SpotPriceResponse{}, fmt.Errorf("no responses found")
+		return WrappedSpotPriceResponse{}, fmt.Errorf("no responses found")
 	}
 
-	perm := rand.Perm(len(responses))
-	for _, i := range perm {
-		resp := responses[perm[i]]
-		if resp.SpotPrice != "" {
-			return resp, nil
+	highestHeight := uint64(0)
+	highestHeightIndex := 0
+
+	for i, resp := range responses {
+		if resp.BlockHeight > highestHeight {
+			highestHeight = resp.BlockHeight
+			highestHeightIndex = i
 		}
 	}
 
-	return SpotPriceResponse{}, fmt.Errorf("no responses found")
+	// 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
 }