fix(wallet): treat unmapped market tokens as non-critical errors

* fall back to the Ethereum sibling for chart history

Author: friofry

internal/circuitbreaker/circuit_breaker.go

@@ -55,9 +55,10 @@ func (cr *CommandResult) addCallStatus(providerName string, err error, startTime
 }
 
 type Command struct {
-	ctx      context.Context
-	functors []*Functor
-	cancel   atomic.Bool
+	ctx                       context.Context
+	functors                  []*Functor
+	cancel                    atomic.Bool
+	nonFailureErrorClassifier func(error) bool
 }
 
 func NewCommand(ctx context.Context, functors []*Functor) *Command {
@@ -79,6 +80,14 @@ func (cmd *Command) Cancel() {
 	cmd.cancel.Store(true)
 }
 
+func (cmd *Command) SetNonFailureErrorClassifier(classifier func(error) bool) {
+	cmd.nonFailureErrorClassifier = classifier
+}
+
+func (cmd *Command) shouldIgnoreFailure(err error) bool {
+	return err != nil && cmd.nonFailureErrorClassifier != nil && cmd.nonFailureErrorClassifier(err)
+}
+
 type Config struct {
 	Timeout                int
 	MaxConcurrentRequests  int
@@ -185,6 +194,7 @@ func (cb *CircuitBreaker) Execute(cmd *Command) CommandResult {
 		}
 
 		var err error
+		var skippedErr error
 		circuitName := f.circuitName
 		providerName := f.providerName
 		if cb.circuitNameHandler != nil {
@@ -231,6 +241,10 @@ func (cb *CircuitBreaker) Execute(cmd *Command) CommandResult {
 					shared.markCancelled()
 					return nil
 				}
+				if cmd.shouldIgnoreFailure(err) {
+					skippedErr = err
+					return nil
+				}
 				if err != nil {
 					logutils.ZapLogger().Warn("hystrix error",
 						zap.String("provider", f.providerName),
@@ -240,6 +254,9 @@ func (cb *CircuitBreaker) Execute(cmd *Command) CommandResult {
 				return err
 			}, nil)
 		}
+		if err == nil && skippedErr != nil {
+			err = skippedErr
+		}
 		if err == nil {
 			break
 		}

internal/circuitbreaker/circuit_breaker_test.go

@@ -647,3 +647,32 @@ func TestCircuitBreaker_ErrorLogging(t *testing.T) {
 		assert.True(t, duration >= 0)
 	})
 }
+
+func TestExecuteIgnoresNonFailureErrorForCircuitAndTriesNextProvider(t *testing.T) {
+	cb := NewCircuitBreaker(Config{
+		Timeout:               1000,
+		MaxConcurrentRequests: 10,
+		SleepWindow:           1000,
+		ErrorPercentThreshold: 1,
+	})
+	mappingErr := errors.New("token not mapped")
+
+	first := NewFunctor(func() ([]any, error) {
+		return nil, mappingErr
+	}, "market-provider-1", "provider-1")
+	second := NewFunctor(func() ([]any, error) {
+		return []any{"ok"}, nil
+	}, "market-provider-2", "provider-2")
+
+	cmd := NewCommand(context.Background(), []*Functor{first, second})
+	cmd.SetNonFailureErrorClassifier(func(err error) bool {
+		return errors.Is(err, mappingErr)
+	})
+
+	result := cb.Execute(cmd)
+	require.NoError(t, result.Error())
+	require.Equal(t, []any{"ok"}, result.Result())
+	require.Len(t, result.FunctorCallStatuses(), 2)
+	require.Error(t, result.FunctorCallStatuses()[0].Err)
+	require.NoError(t, result.FunctorCallStatuses()[1].Err)
+}

internal/healthmanager/provider_errors/provider_errors.go

@@ -26,6 +26,7 @@ const (
 	ProviderErrorTypeNone                    ProviderErrorType = "none"
 	ProviderErrorTypeContextCanceled         ProviderErrorType = "context_canceled"
 	ProviderErrorTypeContextDeadlineExceeded ProviderErrorType = "context_deadline"
+	ProviderErrorTypeDataUnavailable         ProviderErrorType = "data_unavailable"
 	ProviderErrorTypeConnection              ProviderErrorType = "connection"
 	ProviderErrorTypeNotAuthorized           ProviderErrorType = "not_authorized"
 	ProviderErrorTypeForbidden               ProviderErrorType = "forbidden"
@@ -38,6 +39,8 @@ const (
 	ProviderErrorTypeOther                   ProviderErrorType = "other"
 )
 
+var ErrDataUnavailable = errors.New("requested data not available from provider")
+
 // IsConnectionError checks if the error is related to network issues.
 func IsConnectionError(err error) bool {
 	if err == nil {
@@ -225,6 +228,9 @@ func determineProviderErrorType(err error) ProviderErrorType {
 	if errors.Is(err, puzzleauth.ErrAuthRotating) {
 		return ProviderErrorTypeAuthRotating
 	}
+	if errors.Is(err, ErrDataUnavailable) {
+		return ProviderErrorTypeDataUnavailable
+	}
 	if IsConnectionError(err) {
 		return ProviderErrorTypeConnection
 	}
@@ -258,9 +264,20 @@ func IsNonCriticalProviderError(err error) bool {
 	errorType := determineProviderErrorType(err)
 
 	switch errorType {
-	case ProviderErrorTypeNone, ProviderErrorTypeContextCanceled, ProviderErrorTypeContentTooLarge, ProviderErrorTypeRateLimit, ProviderErrorTypeAuthRotating:
+	case ProviderErrorTypeNone, ProviderErrorTypeContextCanceled, ProviderErrorTypeContentTooLarge, ProviderErrorTypeRateLimit, ProviderErrorTypeAuthRotating, ProviderErrorTypeDataUnavailable:
 		return true
 	default:
 		return false
 	}
 }
+
+// IsIgnorableForConnectivity reports whether the error should not flip
+// provider/chain connectivity status to "down".
+func IsIgnorableForConnectivity(err error) bool {
+	if err == nil {
+		return false
+	}
+	return errors.Is(err, context.Canceled) ||
+		IsNonCriticalRpcError(err) ||
+		IsNonCriticalProviderError(err)
+}

internal/healthmanager/provider_errors/provider_errors_test.go

@@ -139,3 +139,33 @@ func TestIsNonCriticalProviderError_PuzzleAuthRotating(t *testing.T) {
 		t.Error("IsNonCriticalProviderError should be false for unrelated errors")
 	}
 }
+
+func TestDetermineProviderErrorType_DataUnavailable(t *testing.T) {
+	if got := determineProviderErrorType(ErrDataUnavailable); got != ProviderErrorTypeDataUnavailable {
+		t.Errorf("determineProviderErrorType(ErrDataUnavailable) = %q, want %q", got, ProviderErrorTypeDataUnavailable)
+	}
+
+	wrapped := fmt.Errorf("wrapped: %w", ErrDataUnavailable)
+	if got := determineProviderErrorType(wrapped); got != ProviderErrorTypeDataUnavailable {
+		t.Errorf("determineProviderErrorType(wrapped ErrDataUnavailable) = %q, want %q", got, ProviderErrorTypeDataUnavailable)
+	}
+}
+
+func TestIsNonCriticalProviderError_DataUnavailable(t *testing.T) {
+	requireErr := fmt.Errorf("wrapped: %w", ErrDataUnavailable)
+	if !IsNonCriticalProviderError(requireErr) {
+		t.Error("IsNonCriticalProviderError should be true for wrapped ErrDataUnavailable")
+	}
+	if !IsNonCriticalProviderError(ErrDataUnavailable) {
+		t.Error("IsNonCriticalProviderError should be true for bare ErrDataUnavailable")
+	}
+}
+
+func TestIsIgnorableForConnectivity(t *testing.T) {
+	if !IsIgnorableForConnectivity(fmt.Errorf("wrapped: %w", ErrDataUnavailable)) {
+		t.Error("IsIgnorableForConnectivity should be true for wrapped ErrDataUnavailable")
+	}
+	if IsIgnorableForConnectivity(errors.New("fatal provider error")) {
+		t.Error("IsIgnorableForConnectivity should be false for unrelated errors")
+	}
+}

services/wallet/collectibles/collectibles_status.go

@@ -1,7 +1,6 @@
 package collectibles
 
 import (
-	"context"
 	"errors"
 	"sync"
 
@@ -86,19 +85,7 @@ func (o *Manager) setChainConnected(chainID walletCommon.ChainID, connected bool
 }
 
 func isCollectiblesIgnorableError(err error) bool {
-	if err == nil {
-		return false
-	}
-	if errors.Is(err, context.Canceled) {
-		return true
-	}
-	if provider_errors.IsNonCriticalRpcError(err) {
-		return true
-	}
-	if provider_errors.IsNonCriticalProviderError(err) {
-		return true
-	}
-	return false
+	return provider_errors.IsIgnorableForConnectivity(err)
 }
 
 func logProviderSearchErr(method, providerID string, chainID walletCommon.ChainID, err error) {

services/wallet/market/market.go

@@ -2,6 +2,7 @@ package market
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"time"
 
@@ -11,6 +12,7 @@ import (
 	"github.com/ethereum/go-ethereum/event"
 
 	"github.com/status-im/status-go/internal/circuitbreaker"
+	provider_errors "github.com/status-im/status-go/internal/healthmanager/provider_errors"
 	"github.com/status-im/status-go/internal/logutils"
 	"github.com/status-im/status-go/services/wallet/thirdparty"
 	tokentypes "github.com/status-im/status-go/services/wallet/token/types"
@@ -100,6 +102,7 @@ func (pm *Manager) setIsConnected(value bool) {
 
 func (pm *Manager) makeCall(providers []thirdparty.MarketDataProvider, f func(provider thirdparty.MarketDataProvider) (interface{}, error)) (interface{}, error) {
 	cmd := circuitbreaker.NewCommand(context.Background(), nil)
+	cmd.SetNonFailureErrorClassifier(provider_errors.IsIgnorableForConnectivity)
 	for _, provider := range providers {
 		provider := provider
 		// FIXME: we might want a different circuitName. See other uses of NewFunctor
@@ -111,16 +114,52 @@ func (pm *Manager) makeCall(providers []thirdparty.MarketDataProvider, f func(pr
 	}
 
 	result := pm.circuitbreaker.Execute(cmd)
-	pm.setIsConnected(result.Error() == nil)
+	isConnectivityError := result.Error() != nil && !provider_errors.IsIgnorableForConnectivity(result.Error())
+	pm.setIsConnected(!isConnectivityError)
 
 	if result.Error() != nil {
-		logutils.ZapLogger().Error("Error fetching prices", zap.Error(result.Error()))
+		if provider_errors.IsIgnorableForConnectivity(result.Error()) {
+			logutils.ZapLogger().Warn("market data unavailable for token mapping", zap.Error(result.Error()))
+		} else {
+			logutils.ZapLogger().Error("Error fetching prices", zap.Error(result.Error()))
+		}
 		return nil, result.Error()
 	}
 
 	return result.Result()[0], nil
 }
 
+func (pm *Manager) fetchHistoricalPricesWithFallback(
+	token *tokentypes.Token,
+	call func(provider thirdparty.MarketDataProvider, token *tokentypes.Token) ([]thirdparty.HistoricalPrice, error),
+) ([]thirdparty.HistoricalPrice, error) {
+	var sibling *tokentypes.Token
+	if allTokens, err := pm.tokenManager.GetTokensForFetchingMarketData(); err == nil {
+		sibling = tokentypes.EthereumMainnetSibling(token, allTokens)
+	}
+
+	for _, t := range []*tokentypes.Token{token, sibling} {
+		if t == nil {
+			continue
+		}
+
+		result, err := pm.makeCall(pm.providers, func(provider thirdparty.MarketDataProvider) (interface{}, error) {
+			return call(provider, t)
+		})
+		if err == nil {
+			return result.([]thirdparty.HistoricalPrice), nil
+		}
+		if !errors.Is(err, thirdparty.ErrTokenNotMapped) {
+			return nil, err
+		}
+	}
+
+	logutils.ZapLogger().Warn("no mapped market history token found",
+		zap.String("tokenKey", token.Key()),
+		zap.String("crossChainID", token.CrossChainID))
+	return []thirdparty.HistoricalPrice{}, nil
+}
+
 func (pm *Manager) getTokensByKeys(tokensKeys []string) (tokens []*tokentypes.Token, err error) {
 	if len(tokensKeys) > 0 {
 		tokens, err = pm.tokenManager.GetTokensByKeysForFetchingMarketData(tokensKeys)
@@ -136,17 +175,9 @@ func (pm *Manager) FetchHistoricalDailyPrices(tokenKey string, currency string,
 		return nil, err
 	}
 
-	result, err := pm.makeCall(pm.providers, func(provider thirdparty.MarketDataProvider) (interface{}, error) {
-		return provider.FetchHistoricalDailyPrices(token, currency, limit, allData, aggregate)
+	return pm.fetchHistoricalPricesWithFallback(token, func(provider thirdparty.MarketDataProvider, t *tokentypes.Token) ([]thirdparty.HistoricalPrice, error) {
+		return provider.FetchHistoricalDailyPrices(t, currency, limit, allData, aggregate)
 	})
-
-	if err != nil {
-		logutils.ZapLogger().Error("Error fetching prices", zap.Error(err))
-		return nil, err
-	}
-
-	prices := result.([]thirdparty.HistoricalPrice)
-	return prices, nil
 }
 
 func (pm *Manager) FetchHistoricalHourlyPrices(tokenKey string, currency string, limit int, aggregate int) ([]thirdparty.HistoricalPrice, error) {
@@ -155,17 +186,9 @@ func (pm *Manager) FetchHistoricalHourlyPrices(tokenKey string, currency string,
 		return nil, err
 	}
 
-	result, err := pm.makeCall(pm.providers, func(provider thirdparty.MarketDataProvider) (interface{}, error) {
-		return provider.FetchHistoricalHourlyPrices(token, currency, limit, aggregate)
+	return pm.fetchHistoricalPricesWithFallback(token, func(provider thirdparty.MarketDataProvider, t *tokentypes.Token) ([]thirdparty.HistoricalPrice, error) {
+		return provider.FetchHistoricalHourlyPrices(t, currency, limit, aggregate)
 	})
-
-	if err != nil {
-		logutils.ZapLogger().Error("Error fetching prices", zap.Error(err))
-		return nil, err
-	}
-
-	prices := result.([]thirdparty.HistoricalPrice)
-	return prices, nil
 }
 
 // FetchTokenMarketValues fetches market values for a given token keys and currency. If no tokens are provided, all tokens are fetched.

services/wallet/market/market_feed_test.go

@@ -50,11 +50,8 @@ func (s *MarketTestSuite) TestEventOnRpsError() {
 	// WHEN
 	_, err := manager.FetchPrices(s.tokensKeys, s.currencies)
 	s.Require().Error(err, "expected error from FetchPrices due to MockPriceProviderWithError")
-	event, ok := s.feedSub.WaitForEvent(5 * time.Second)
-	s.Require().True(ok, "expected an event, but none was received")
-
-	// THEN
-	s.Require().Equal(event.Type, EventMarketStatusChanged)
+	_, ok := s.feedSub.WaitForEvent(500 * time.Millisecond)
+	s.Require().False(ok, "expected no status event for non-critical rate limit error")
 }
 
 func (s *MarketTestSuite) TestEventOnNetworkError() {