feat: Reduce blast radius of errors in MarketMap (#736)

oracle/helpers_test.go

@@ -28,6 +28,8 @@ import (
 
 var (
 	btcusdtCP = slinkytypes.NewCurrencyPair("BTC", "USDT")
+	btcusdCP  = slinkytypes.NewCurrencyPair("BTC", "USD")
+	usdtusdCP = slinkytypes.NewCurrencyPair("USDT", "USD")
 	ethusdtCP = slinkytypes.NewCurrencyPair("ETH", "USDT")
 )
 
@@ -166,6 +168,72 @@ var (
 		Type: mmclienttypes.ConfigType,
 	}
 
+	validMarketMapSubset = mmtypes.MarketMap{
+		Markets: map[string]mmtypes.Market{
+			ethusdtCP.String(): {
+				Ticker: mmtypes.Ticker{
+					CurrencyPair:     ethusdtCP,
+					MinProviderCount: 1,
+					Decimals:         8,
+					Enabled:          true,
+				},
+				ProviderConfigs: []mmtypes.ProviderConfig{
+					{
+						Name:           coinbase.Name,
+						OffChainTicker: coinbaseethusd.GetOffChainTicker(),
+					},
+					{
+						Name:           okx.Name,
+						OffChainTicker: okxethusd.GetOffChainTicker(),
+					},
+				},
+			},
+		},
+	}
+
+	partialInvalidMarketMap = mmtypes.MarketMap{
+		Markets: map[string]mmtypes.Market{
+			btcusdCP.String(): {
+				Ticker: mmtypes.Ticker{
+					CurrencyPair:     btcusdCP,
+					MinProviderCount: 1,
+					Decimals:         8,
+					Enabled:          true,
+				},
+				ProviderConfigs: []mmtypes.ProviderConfig{
+					{
+						Name:            coinbase.Name,
+						OffChainTicker:  coinbasebtcusd.GetOffChainTicker(),
+						NormalizeByPair: &usdtusdCP,
+					},
+					{
+						Name:            okx.Name,
+						OffChainTicker:  okxbtcusd.GetOffChainTicker(),
+						NormalizeByPair: &usdtusdCP,
+					},
+				},
+			},
+			ethusdtCP.String(): {
+				Ticker: mmtypes.Ticker{
+					CurrencyPair:     ethusdtCP,
+					MinProviderCount: 1,
+					Decimals:         8,
+					Enabled:          true,
+				},
+				ProviderConfigs: []mmtypes.ProviderConfig{
+					{
+						Name:           coinbase.Name,
+						OffChainTicker: coinbaseethusd.GetOffChainTicker(),
+					},
+					{
+						Name:           okx.Name,
+						OffChainTicker: okxethusd.GetOffChainTicker(),
+					},
+				},
+			},
+		},
+	}
+
 	// Coinbase and OKX are supported by the marketmap.
 	marketMap = mmtypes.MarketMap{
 		Markets: map[string]mmtypes.Market{

oracle/update.go

@@ -19,14 +19,19 @@ func (o *OracleImpl) UpdateMarketMap(marketMap mmtypes.MarketMap) error {
 	o.mut.Lock()
 	defer o.mut.Unlock()
 
-	if err := marketMap.ValidateBasic(); err != nil {
+	validSubset, err := marketMap.GetValidSubset()
+	if err != nil {
 		o.logger.Error("failed to validate market map", zap.Error(err))
 		return err
 	}
 
+	if len(validSubset.Markets) == 0 {
+		o.logger.Warn("market map update produced no valid markets to fetch")
+	}
+
 	// Iterate over all existing price providers and update their market maps.
 	for name, state := range o.priceProviders {
-		providerTickers, err := types.ProviderTickersFromMarketMap(name, marketMap)
+		providerTickers, err := types.ProviderTickersFromMarketMap(name, validSubset)
 		if err != nil {
 			o.logger.Error("failed to create provider market map", zap.String("provider", name), zap.Error(err))
 			return err
@@ -42,7 +47,7 @@ func (o *OracleImpl) UpdateMarketMap(marketMap mmtypes.MarketMap) error {
 		o.priceProviders[name] = updatedState
 	}
 
-	o.marketMap = marketMap
+	o.marketMap = validSubset
 	if o.aggregator != nil {
 		o.aggregator.UpdateMarketMap(o.marketMap)
 	}

oracle/update_test.go

@@ -18,7 +18,7 @@ import (
 )
 
 func TestUpdateWithMarketMap(t *testing.T) {
-	t.Run("bad market map is rejected", func(t *testing.T) {
+	t.Run("bad market map is not rejected", func(t *testing.T) {
 		orc, err := oracle.New(
 			oracleCfg,
 			noOpPriceAggregator{},
@@ -35,7 +35,7 @@ func TestUpdateWithMarketMap(t *testing.T) {
 				"bad": {},
 			},
 		})
-		require.Error(t, err)
+		require.NoError(t, err)
 
 		o.Stop()
 	})
@@ -626,4 +626,60 @@ func TestUpdateProviderState(t *testing.T) {
 			500*time.Millisecond,
 		)
 	})
+
+	t.Run("can update the market map with partial failure on NormalizeBy", func(t *testing.T) {
+		orc, err := oracle.New(
+			oracleCfg,
+			noOpPriceAggregator{},
+			oracle.WithLogger(logger),
+			oracle.WithPriceAPIQueryHandlerFactory(oraclefactory.APIQueryHandlerFactory),
+			oracle.WithPriceWebSocketQueryHandlerFactory(oraclefactory.WebSocketQueryHandlerFactory),
+		)
+		require.NoError(t, err)
+		o := orc.(*oracle.OracleImpl)
+		require.NoError(t, o.Init(context.TODO()))
+
+		providers := o.GetProviderState()
+		require.Len(t, providers, 3)
+
+		// Update the oracle's market map.
+		require.NoError(t, o.UpdateMarketMap(partialInvalidMarketMap))
+
+		providers = o.GetProviderState()
+
+		cbTickers, err := types.ProviderTickersFromMarketMap(coinbase.Name, validMarketMapSubset)
+		require.NoError(t, err)
+
+		// Check the state after the update.
+		coinbaseState, ok := providers[coinbase.Name]
+		require.True(t, ok)
+		checkProviderState(
+			t,
+			cbTickers,
+			coinbase.Name,
+			providertypes.API,
+			false,
+			coinbaseState,
+		)
+
+		okxTickers, err := types.ProviderTickersFromMarketMap(okx.Name, validMarketMapSubset)
+		require.NoError(t, err)
+
+		okxState, ok := providers[okx.Name]
+		require.True(t, ok)
+		checkProviderState(
+			t,
+			okxTickers,
+			okx.Name,
+			providertypes.WebSockets,
+			false,
+			okxState,
+		)
+
+		binanceState, ok := providers[binance.Name]
+		require.True(t, ok)
+		checkProviderState(t, nil, binance.Name, providertypes.API, false, binanceState)
+
+		o.Stop()
+	})
 }

x/marketmap/types/market.go

@@ -40,6 +40,52 @@ func (mm *MarketMap) ValidateBasic() error {
 	return nil
 }
 
+// GetValidSubset outputs a MarketMap which contains the maximal valid subset of this MarketMap.
+//
+//	In particular, this will eliminate anything which would otherwise cause a failure in ValidateBasic.
+//	The resulting MarketMap should be able to pass ValidateBasic.
+func (mm *MarketMap) GetValidSubset() (MarketMap, error) {
+	validSubset := MarketMap{Markets: make(map[string]Market)}
+
+	// Operates in 2 passes:
+	// 1. Remove invalid ProviderConfigs
+	for ticker, market := range mm.Markets {
+		var validProviderConfigs []ProviderConfig
+		for _, providerConfig := range market.ProviderConfigs {
+			if providerConfig.NormalizeByPair != nil {
+				normalizeMarket, found := mm.Markets[providerConfig.NormalizeByPair.String()]
+				if !found {
+					continue
+				}
+
+				if !normalizeMarket.Ticker.Enabled && market.Ticker.Enabled {
+					continue
+				}
+			}
+			validProviderConfigs = append(validProviderConfigs, providerConfig)
+		}
+		market.ProviderConfigs = validProviderConfigs
+		validSubset.Markets[ticker] = market
+	}
+	// 2. Remove ValidateBasic failures on all included markets
+	for ticker, market := range validSubset.Markets {
+		if err := market.ValidateBasic(); err != nil {
+			delete(validSubset.Markets, ticker)
+			continue
+		}
+		// expect that the ticker (index) is equal to the market.Ticker.String()
+		if ticker != market.Ticker.String() {
+			delete(validSubset.Markets, ticker)
+			continue
+		}
+	}
+	if valErr := validSubset.ValidateBasic(); valErr != nil {
+		return validSubset, valErr
+	}
+
+	return validSubset, nil
+}
+
 // String returns the string representation of the market map.
 func (mm *MarketMap) String() string {
 	return fmt.Sprintf(