Configurable f3 nodes (#540)

* feat: update downstream parsers to use a list of nodes

* feat: update actors that use cache to use a list of nodes

* feat: update actors cache and helper to use list of nodes

* feat: update factory and v1 and v2 parsers to use a list of nodes

* test: update tests

* test: update mock expectations

* fix: retain cache setup name

* fix: use context to access nodes

* fix: use context to access nodes

* fix: node loop logic

* feat: pass nodes to downstream parsers

Author: ziscky

actors/address.go

@@ -1,6 +1,7 @@
 package actors
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/filecoin-project/go-address"
@@ -13,7 +14,7 @@ import (
 // if the address is a zero address account actor, it returns the robust address of the zero address account actor
 // if the address is already a robust address, it returns the address
 // if the address is f2 evm, we consolidate f2 -> f0 -> f4
-func ConsolidateToRobustAddress(addr address.Address, h *helper.Helper, logger *logger.Logger, bestEffort bool, canonical bool) (string, error) {
+func ConsolidateToRobustAddress(ctx context.Context, addr address.Address, h *helper.Helper, logger *logger.Logger, bestEffort bool, canonical bool) (string, error) {
 	actorCache := h.GetActorsCache()
 	if ok, _, _ := h.IsZeroAddressAccountActor(addr); ok {
 		return helper.ZeroAddressAccountActorRobust, nil
@@ -22,18 +23,18 @@ func ConsolidateToRobustAddress(addr address.Address, h *helper.Helper, logger *
 	if isRobust, _ := common.IsRobustAddress(addr); isRobust {
 		// we need to handle cases where a f2 address for evm actors is used
 		// f2 -> f0 -> f4, as we want to consolidate the address to f4 style
-		shortAddressStr, err := actorCache.GetShortAddress(addr, canonical)
+		shortAddressStr, err := actorCache.GetShortAddress(ctx, addr, canonical)
 		if err == nil {
 			shortAddress, _ := address.NewFromString(shortAddressStr)
-			addrStr, err := actorCache.GetRobustAddress(shortAddress, canonical)
+			addrStr, err := actorCache.GetRobustAddress(ctx, shortAddress, canonical)
 			if err == nil {
 				addr, _ = address.NewFromString(addrStr)
 			}
 		}
 		return addr.String(), nil
 	}
 
-	robustAddress, err := actorCache.GetRobustAddress(addr, canonical)
+	robustAddress, err := actorCache.GetRobustAddress(ctx, addr, canonical)
 	if err != nil && !bestEffort {
 		logger.Warnf("Error converting address %s to robust format: %v", addr, err)
 		return "", fmt.Errorf("error converting address to robust format: %v", err) // Fallback

actors/cache/actors_cache.go

@@ -126,10 +126,10 @@ func (a *ActorsCache) ClearBadAddressCache() {
 	a.badAddress.Clear()
 }
 
-func (a *ActorsCache) GetActorCode(add address.Address, key filTypes.TipSetKey, onChainOnly, canonical bool) (string, error) {
+func (a *ActorsCache) GetActorCode(ctx context.Context, add address.Address, key filTypes.TipSetKey, onChainOnly, canonical bool) (string, error) {
 	addStr := add.String()
 
-	store, actorCode, err := a.getActorCode(add, key, onChainOnly, canonical)
+	store, actorCode, err := a.getActorCode(ctx, add, key, onChainOnly, canonical)
 	if err != nil {
 		a.logger.Errorf("[ActorsCache] - Unable to retrieve actor code from node: %s", err.Error())
 		if strings.Contains(err.Error(), "actor not found") {
@@ -144,7 +144,7 @@ func (a *ActorsCache) GetActorCode(add address.Address, key filTypes.TipSetKey,
 	}
 
 	// Code is not cached, store it
-	err = a.storeActorCode(add, types.AddressInfo{
+	err = a.storeActorCode(ctx, add, types.AddressInfo{
 		ActorCid:    actorCode,
 		IsCanonical: canonical,
 	})
@@ -157,8 +157,8 @@ func (a *ActorsCache) GetActorCode(add address.Address, key filTypes.TipSetKey,
 	return actorCode, nil
 }
 
-func (a *ActorsCache) GetRobustAddress(add address.Address, canonical bool) (string, error) {
-	store, robust, err := a.getRobustAddress(add, canonical)
+func (a *ActorsCache) GetRobustAddress(ctx context.Context, add address.Address, canonical bool) (string, error) {
+	store, robust, err := a.getRobustAddress(ctx, add, canonical)
 	if err != nil {
 		return "", err
 	}
@@ -168,7 +168,7 @@ func (a *ActorsCache) GetRobustAddress(add address.Address, canonical bool) (str
 	}
 
 	// Robust address is not cached, store it
-	err = a.storeRobustAddress(add, types.AddressInfo{
+	err = a.storeRobustAddress(ctx, add, types.AddressInfo{
 		Robust:      robust,
 		IsCanonical: canonical,
 	})
@@ -181,8 +181,8 @@ func (a *ActorsCache) GetRobustAddress(add address.Address, canonical bool) (str
 	return robust, nil
 }
 
-func (a *ActorsCache) GetShortAddress(add address.Address, canonical bool) (string, error) {
-	store, short, err := a.getShortAddress(add, canonical)
+func (a *ActorsCache) GetShortAddress(ctx context.Context, add address.Address, canonical bool) (string, error) {
+	store, short, err := a.getShortAddress(ctx, add, canonical)
 	if err != nil {
 		return "", err
 	}
@@ -191,7 +191,7 @@ func (a *ActorsCache) GetShortAddress(add address.Address, canonical bool) (stri
 		return short, nil
 	}
 	// Robust address is not cached, store it
-	err = a.storeShortAddress(add, types.AddressInfo{
+	err = a.storeShortAddress(ctx, add, types.AddressInfo{
 		Short:       short,
 		IsCanonical: canonical,
 	})
@@ -242,10 +242,10 @@ func (a *ActorsCache) getEVMSelectorSig(ctx context.Context, selectorID string,
 	return "", err
 }
 
-func (a *ActorsCache) getShortAddress(add address.Address, canonical bool) (store bool, shortAddress string, err error) {
+func (a *ActorsCache) getShortAddress(ctx context.Context, add address.Address, canonical bool) (store bool, shortAddress string, err error) {
 	addStr := add.String()
 	// Try canonical cache
-	short, err := a.offChainCache.GetShortAddress(add, canonical)
+	short, err := a.offChainCache.GetShortAddress(ctx, add, canonical)
 	if err == nil {
 		return false, short, nil
 	}
@@ -255,7 +255,7 @@ func (a *ActorsCache) getShortAddress(add address.Address, canonical bool) (stor
 	if a.isBadAddress(add) {
 		return false, "", fmt.Errorf("address %s is flagged as bad", addStr)
 	}
-	short, err = a.onChainCache.GetShortAddress(add, canonical)
+	short, err = a.onChainCache.GetShortAddress(ctx, add, canonical)
 	if err != nil {
 		a.logger.Debugf("[ActorsCache] - Unable to retrieve short address from onchain cache for address %s.", addStr)
 		return false, "", err
@@ -264,7 +264,7 @@ func (a *ActorsCache) getShortAddress(add address.Address, canonical bool) (stor
 	return true, short, nil
 }
 
-func (a *ActorsCache) getRobustAddress(add address.Address, canonical bool) (store bool, robustAddr string, err error) {
+func (a *ActorsCache) getRobustAddress(ctx context.Context, add address.Address, canonical bool) (store bool, robustAddr string, err error) {
 	addStr := add.String()
 	// check if the address is a system actor ( no robust address)
 	if _, ok := SystemActorsId[addStr]; ok {
@@ -281,7 +281,7 @@ func (a *ActorsCache) getRobustAddress(add address.Address, canonical bool) (sto
 		}
 	}
 
-	robust, err := a.offChainCache.GetRobustAddress(add, canonical)
+	robust, err := a.offChainCache.GetRobustAddress(ctx, add, canonical)
 	if err == nil {
 		return false, robust, nil
 	}
@@ -290,7 +290,7 @@ func (a *ActorsCache) getRobustAddress(add address.Address, canonical bool) (sto
 	if a.isBadAddress(add) {
 		return false, "", fmt.Errorf("%w: address %s is flagged as bad", ErrBadAddress, addStr)
 	}
-	robust, err = a.onChainCache.GetRobustAddress(add, canonical)
+	robust, err = a.onChainCache.GetRobustAddress(ctx, add, canonical)
 	if err != nil {
 		a.logger.Debugf("[ActorsCache] - Unable to retrieve robust address from onchain cache for address %s.", addStr)
 		return false, "", err
@@ -299,9 +299,9 @@ func (a *ActorsCache) getRobustAddress(add address.Address, canonical bool) (sto
 	return true, robust, nil
 }
 
-func (a *ActorsCache) getActorCode(add address.Address, key filTypes.TipSetKey, onChainOnly, canonical bool) (store bool, actorCode string, err error) {
+func (a *ActorsCache) getActorCode(ctx context.Context, add address.Address, key filTypes.TipSetKey, onChainOnly, canonical bool) (store bool, actorCode string, err error) {
 	addStr := add.String()
-	actorCode, err = a.offChainCache.GetActorCode(add, key, onChainOnly, canonical)
+	actorCode, err = a.offChainCache.GetActorCode(ctx, add, key, onChainOnly, canonical)
 	if err == nil {
 		return false, actorCode, nil
 	}
@@ -311,7 +311,7 @@ func (a *ActorsCache) getActorCode(add address.Address, key filTypes.TipSetKey,
 		return false, "", fmt.Errorf(" %w : %s is flagged as bad", ErrBadAddress, addStr)
 	}
 
-	actorCode, err = a.onChainCache.GetActorCode(add, key, onChainOnly, canonical)
+	actorCode, err = a.onChainCache.GetActorCode(ctx, add, key, onChainOnly, canonical)
 	if err != nil {
 		a.logger.Debugf("[ActorsCache] - Unable to retrieve actor code from onchain cache for address %s.", addStr)
 		return false, "", err
@@ -324,8 +324,8 @@ func (a *ActorsCache) StoreEVMSelectorSig(ctx context.Context, selectorID string
 	return a.offChainCache.StoreEVMSelectorSig(ctx, selectorID, sig, canonical)
 }
 
-func (a *ActorsCache) storeActorCode(add address.Address, info types.AddressInfo) error {
-	shortAddress, err := a.GetShortAddress(add, info.IsCanonical)
+func (a *ActorsCache) storeActorCode(ctx context.Context, add address.Address, info types.AddressInfo) error {
+	shortAddress, err := a.GetShortAddress(ctx, add, info.IsCanonical)
 	if err != nil {
 		return err
 	}
@@ -339,8 +339,8 @@ func (a *ActorsCache) storeActorCode(add address.Address, info types.AddressInfo
 	return nil
 }
 
-func (a *ActorsCache) storeShortAddress(add address.Address, info types.AddressInfo) error {
-	_, robustAddress, err := a.getRobustAddress(add, info.IsCanonical)
+func (a *ActorsCache) storeShortAddress(ctx context.Context, add address.Address, info types.AddressInfo) error {
+	_, robustAddress, err := a.getRobustAddress(ctx, add, info.IsCanonical)
 	if err != nil {
 		return err
 	}
@@ -354,8 +354,8 @@ func (a *ActorsCache) storeShortAddress(add address.Address, info types.AddressI
 	return nil
 }
 
-func (a *ActorsCache) storeRobustAddress(add address.Address, info types.AddressInfo) error {
-	_, shortAddress, err := a.getShortAddress(add, info.IsCanonical)
+func (a *ActorsCache) storeRobustAddress(ctx context.Context, add address.Address, info types.AddressInfo) error {
+	_, shortAddress, err := a.getShortAddress(ctx, add, info.IsCanonical)
 	if err != nil {
 		return err
 	}

actors/cache/impl/onchain.go

@@ -16,6 +16,8 @@ import (
 	filTypes "github.com/filecoin-project/lotus/chain/types"
 	"github.com/ipfs/go-cid"
 	"github.com/zondax/fil-parser/actors/cache/impl/common"
+	parserContext "github.com/zondax/fil-parser/context"
+
 	golemBackoff "github.com/zondax/golem/pkg/zhttpclient/backoff"
 
 	cacheMetrics "github.com/zondax/fil-parser/actors/cache/metrics"
@@ -27,7 +29,6 @@ const OnChainImpl = "on-chain"
 
 // OnChain implementation
 type OnChain struct {
-	Node       api.FullNode
 	logger     *logger.Logger
 	backoff    *golemBackoff.BackOff
 	metrics    *cacheMetrics.ActorsCacheMetricsClient
@@ -46,11 +47,7 @@ func (m *OnChain) BackFill() error {
 func (m *OnChain) NewImpl(source common.DataSource, logger *logger.Logger, metrics *cacheMetrics.ActorsCacheMetricsClient, backoff *golemBackoff.BackOff) error {
 	// Node datastore is required
 	m.logger = logger2.GetSafeLogger(logger)
-	if source.Node == nil {
-		m.logger.Panic("[ActorsCache] - Node ptr is nil")
-	}
 
-	m.Node = source.Node
 	m.metrics = metrics
 	m.backoff = backoff
 	m.httpClient = resty.New().SetTimeout(30 * time.Second)
@@ -62,16 +59,30 @@ func (m *OnChain) ImplementationType() string {
 	return OnChainImpl
 }
 
-func (m *OnChain) GetActorCode(address address.Address, key filTypes.TipSetKey, _, _ bool) (string, error) {
-	actorCid, err := m.retrieveActorFromLotus(address, key)
+func (m *OnChain) GetActorCode(ctx context.Context, address address.Address, key filTypes.TipSetKey, _, _ bool) (string, error) {
+	var actorCid cid.Cid
+	var err error
+
+	nodes, err := parserContext.GetNodes(ctx)
 	if err != nil {
-		return cid.Undef.String(), err
+		return "", err
+	}
+	if len(nodes) == 0 {
+		return "", fmt.Errorf("no nodes available")
+	}
+	for _, node := range nodes {
+		actorCid, err = m.retrieveActorFromLotus(node, address, key)
+		if err == nil {
+			break
+		} else if !IsRetriableError(err) {
+			return cid.Undef.String(), err
+		}
 	}
 
 	return actorCid.String(), nil
 }
 
-func (m *OnChain) GetRobustAddress(address address.Address, _ bool) (string, error) {
+func (m *OnChain) GetRobustAddress(ctx context.Context, address address.Address, _ bool) (string, error) {
 	isRobustAddress, err := common.IsRobustAddress(address)
 	if err != nil {
 		return "", err
@@ -82,16 +93,28 @@ func (m *OnChain) GetRobustAddress(address address.Address, _ bool) (string, err
 		return address.String(), nil
 	}
 
-	// Address is not in cache, get robust address from lotus
-	robustAdd, err := m.retrieveActorPubKeyFromLotus(address, false)
+	var robustAdd string
+	nodes, err := parserContext.GetNodes(ctx)
 	if err != nil {
 		return "", err
 	}
+	if len(nodes) == 0 {
+		return "", fmt.Errorf("no nodes available")
+	}
+	// Address is not in cache, get robust address from lotus
+	for _, node := range nodes {
+		robustAdd, err = m.retrieveActorPubKeyFromLotus(node, address, false)
+		if err == nil {
+			break
+		} else if !IsRetriableError(err) {
+			return "", err
+		}
+	}
 
 	return robustAdd, nil
 }
 
-func (m *OnChain) GetShortAddress(address address.Address, _ bool) (string, error) {
+func (m *OnChain) GetShortAddress(ctx context.Context, address address.Address, _ bool) (string, error) {
 	isRobustAddress, err := common.IsRobustAddress(address)
 	if err != nil {
 		return "", err
@@ -102,9 +125,21 @@ func (m *OnChain) GetShortAddress(address address.Address, _ bool) (string, erro
 		return address.String(), nil
 	}
 
-	shortAdd, err := m.retrieveActorPubKeyFromLotus(address, true)
+	var shortAdd string
+	nodes, err := parserContext.GetNodes(ctx)
 	if err != nil {
-		return "", common.ErrKeyNotFound
+		return "", err
+	}
+	if len(nodes) == 0 {
+		return "", fmt.Errorf("no nodes available")
+	}
+	for _, node := range nodes {
+		shortAdd, err = m.retrieveActorPubKeyFromLotus(node, address, true)
+		if err == nil {
+			break
+		} else if !IsRetriableError(err) {
+			return "", common.ErrKeyNotFound
+		}
 	}
 
 	return shortAdd, nil
@@ -124,12 +159,12 @@ func (m *OnChain) IsGenesisActor(_ string) bool {
 	return false
 }
 
-func (m *OnChain) retrieveActorFromLotus(add address.Address, key filTypes.TipSetKey) (cid.Cid, error) {
+func (m *OnChain) retrieveActorFromLotus(node api.FullNode, add address.Address, key filTypes.TipSetKey) (cid.Cid, error) {
 	nodeApiCallOptions := &NodeApiCallWithRetryOptions[*filTypes.Actor]{
 		RequestName: "StateGetActorWithTipSetKey",
 		BackOff:     *m.backoff,
 		Request: func() (*filTypes.Actor, error) {
-			return m.Node.StateGetActor(context.Background(), add, key)
+			return node.StateGetActor(context.Background(), add, key)
 		},
 		RetryErrStrings: []string{"ipld: could not find", "RPC client error", "503"},
 	}
@@ -139,7 +174,7 @@ func (m *OnChain) retrieveActorFromLotus(add address.Address, key filTypes.TipSe
 		// Try again but with an empty tipset Key
 		nodeApiCallOptions.RequestName = "StateGetActor"
 		nodeApiCallOptions.Request = func() (*filTypes.Actor, error) {
-			return m.Node.StateGetActor(context.Background(), add, filTypes.EmptyTSK)
+			return node.StateGetActor(context.Background(), add, filTypes.EmptyTSK)
 		}
 		actor, err = NodeApiCallWithRetry(nodeApiCallOptions, m.metrics)
 		if err != nil {
@@ -151,7 +186,7 @@ func (m *OnChain) retrieveActorFromLotus(add address.Address, key filTypes.TipSe
 	return actor.Code, nil
 }
 
-func (m *OnChain) retrieveActorPubKeyFromLotus(add address.Address, reverse bool) (string, error) {
+func (m *OnChain) retrieveActorPubKeyFromLotus(node api.FullNode, add address.Address, reverse bool) (string, error) {
 	var key address.Address
 	var err error
 
@@ -163,13 +198,13 @@ func (m *OnChain) retrieveActorPubKeyFromLotus(add address.Address, reverse bool
 	if reverse {
 		nodeApiCallOptions.RequestName = "StateLookupID"
 		nodeApiCallOptions.Request = func() (address.Address, error) {
-			return m.Node.StateLookupID(context.Background(), add, filTypes.EmptyTSK)
+			return node.StateLookupID(context.Background(), add, filTypes.EmptyTSK)
 		}
 		key, err = NodeApiCallWithRetry(nodeApiCallOptions, m.metrics)
 	} else {
 		nodeApiCallOptions.RequestName = "StateAccountKey"
 		nodeApiCallOptions.Request = func() (address.Address, error) {
-			return m.Node.StateAccountKey(context.Background(), add, filTypes.EmptyTSK)
+			return node.StateAccountKey(context.Background(), add, filTypes.EmptyTSK)
 		}
 		key, err = NodeApiCallWithRetry(nodeApiCallOptions, m.metrics)
 	}
@@ -178,7 +213,7 @@ func (m *OnChain) retrieveActorPubKeyFromLotus(add address.Address, reverse bool
 		if strings.Contains(err.Error(), "actor code is not account") {
 			nodeApiCallOptions.RequestName = "StateLookupRobustAddress"
 			nodeApiCallOptions.Request = func() (address.Address, error) {
-				return m.Node.StateLookupRobustAddress(context.Background(), add, filTypes.EmptyTSK)
+				return node.StateLookupRobustAddress(context.Background(), add, filTypes.EmptyTSK)
 			}
 			key, err = NodeApiCallWithRetry(nodeApiCallOptions, m.metrics)
 			if err != nil {