fix(collectibles): reduce redundant nft fetches (#7392)

* check context cancellation on every iteration

fixes status-im/status-app#20307

Author: friofry

services/wallet/collectibles/ownership/controller.go

@@ -466,28 +466,28 @@ func (c *Controller) loadWithPeriodicalLoaderIfFound(ctx context.Context, chainI
 		_ = loader.Load(ctx)
 	}()
 
-	select {
-	case <-ctx.Done():
-		err = ctx.Err()
-		return
-	case finishedEvent, ok := <-finishedCh:
-		if !ok {
-			return
-		}
-		if finishedEvent.ChainID == chainID && finishedEvent.Account == account {
-			return
-		}
-	case errEvent, ok := <-errCh:
-		if !ok {
-			return
-		}
-		if errEvent.ChainID == chainID && errEvent.Account == account {
-			err = errEvent.Error
+	for {
+		select {
+		case <-ctx.Done():
+			err = ctx.Err()
 			return
+		case finishedEvent, ok := <-finishedCh:
+			if !ok {
+				return
+			}
+			if finishedEvent.ChainID == chainID && finishedEvent.Account == account {
+				return
+			}
+		case errEvent, ok := <-errCh:
+			if !ok {
+				return
+			}
+			if errEvent.ChainID == chainID && errEvent.Account == account {
+				err = errEvent.Error
+				return
+			}
 		}
 	}
-
-	return
 }
 
 func (c *Controller) checkPeriodicalLoaders() {

services/wallet/collectibles/ownership/controller_test.go

@@ -3,12 +3,17 @@ package ownership_test
 import (
 	"context"
 	"math/big"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	coretypes "github.com/ethereum/go-ethereum/core/types"
 
 	"github.com/status-im/go-wallet-sdk/pkg/balance/multistandardfetcher"
+	"github.com/status-im/go-wallet-sdk/pkg/contracts/erc1155"
+	"github.com/status-im/go-wallet-sdk/pkg/contracts/erc721"
+	"github.com/status-im/go-wallet-sdk/pkg/eventlog"
 
 	"github.com/stretchr/testify/require"
 
@@ -25,6 +30,7 @@ import (
 	walletCommon "github.com/status-im/status-go/services/wallet/common"
 	"github.com/status-im/status-go/services/wallet/multistandardbalance"
 	"github.com/status-im/status-go/services/wallet/thirdparty"
+	"github.com/status-im/status-go/services/wallet/transferdetector"
 
 	"go.uber.org/mock/gomock"
 	"go.uber.org/zap"
@@ -805,3 +811,128 @@ func TestControllerPeriodicalLoads(t *testing.T) {
 
 	controller.Stop()
 }
+
+func TestControllerTransferDetectionEvents(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	accountsProvider := mock_ownership.NewMockAccountsProvider(mockCtrl)
+	fakeAddress := types.HexToAddress("0x123")
+	accountsProvider.EXPECT().GetWalletAddresses().Return([]types.Address{fakeAddress}, nil).AnyTimes()
+
+	accountsPublisher := pubsub.NewPublisher()
+	networksProvider := mock_ownership.NewMockNetworksProvider(mockCtrl)
+	networksPublisher := pubsub.NewPublisher()
+
+	networksProvider.EXPECT().GetActiveNetworks().Return([]*params.Network{
+		{ChainID: 1, IsActive: true},
+	}, nil).AnyTimes()
+	networksProvider.EXPECT().GetPublisher().Return(networksPublisher).AnyTimes()
+
+	walletDB, err := testutils.SetupTestMemorySQLDB(walletdatabase.DbInitializer{})
+	require.NoError(t, err)
+
+	emptyCollectiblesContainer := &thirdparty.CollectibleOwnershipContainer{
+		Items:          []thirdparty.CollectibleIDBalance{},
+		NextCursor:     "",
+		PreviousCursor: "",
+		Provider:       "mockProvider",
+	}
+	ownershipFetcher := mock_ownership.NewMockCollectibleOwnershipFetcher(mockCtrl)
+	ownershipFetcher.EXPECT().FetchCollectibleOwnershipByOwner(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, chainID walletCommon.ChainID, owner common.Address, cursor string, limit int, providerID string) (*thirdparty.CollectibleOwnershipContainer, error) {
+			return emptyCollectiblesContainer, nil
+		}).AnyTimes()
+
+	multistandardBalancePublisher := pubsub.NewPublisher()
+	transferDetectorPublisher := pubsub.NewPublisher()
+	blockChainStateProvider := mock_ownership.NewMockBlockChainStateProvider(mockCtrl)
+	var estimatedBlockCalls atomic.Int32
+	blockChainStateProvider.EXPECT().GetEstimatedBlockTime(gomock.Any(), uint64(1), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, chainID uint64, blockNumber uint64) (time.Time, error) {
+			estimatedBlockCalls.Add(1)
+			return time.Now(), nil
+		},
+	).AnyTimes()
+
+	publisher := pubsub.NewPublisher()
+	logger := zaptest.NewLogger(t).WithOptions(zap.AddCallerSkip(1))
+
+	controller := ownership.NewController(
+		ownership.NewOwnershipDB(walletDB),
+		accountsProvider,
+		accountsPublisher,
+		networksProvider,
+		multistandardBalancePublisher,
+		transferDetectorPublisher,
+		blockChainStateProvider,
+		ownershipFetcher,
+		publisher,
+		logger,
+	)
+
+	params := ownership.PeriodicalLoaderParams{
+		StartDelay:   1 * time.Second,
+		LoadInterval: 10 * time.Second,
+		LoaderParams: ownership.LoaderParams{
+			LoadDelay:  0 * time.Second,
+			FetchLimit: 50,
+		},
+	}
+	controller.StartWithLoaderParams(params)
+	defer controller.Stop()
+
+	// Ensure initial load completed so ownership timestamp exists.
+	require.Eventually(t, func() bool {
+		return controller.GetLoaderState(walletCommon.ChainID(1), common.Address(fakeAddress)) == ownership.LoaderStateIdle
+	}, 2*time.Second, 50*time.Millisecond)
+
+	otherAddress := common.HexToAddress("0x456")
+	msg := transferdetector.EventTransferDetectionFinished{
+		ChainID:   1,
+		Accounts:  []common.Address{common.Address(fakeAddress)},
+		FromBlock: 100,
+		ToBlock:   200,
+		Events: []eventlog.Event{
+			{
+				EventKey: eventlog.ERC721Transfer,
+				Unpacked: erc721.Erc721Transfer{
+					From: common.Address(fakeAddress),
+					To:   otherAddress,
+					Raw: coretypes.Log{
+						BlockNumber: 123,
+					},
+				},
+			},
+			{
+				EventKey: eventlog.ERC1155TransferSingle,
+				Unpacked: erc1155.Erc1155TransferSingle{
+					From: common.Address(fakeAddress),
+					To:   otherAddress,
+					Raw: coretypes.Log{
+						BlockNumber: 124,
+					},
+				},
+			},
+			{
+				EventKey: eventlog.ERC1155TransferBatch,
+				Unpacked: erc1155.Erc1155TransferBatch{
+					From: common.Address(fakeAddress),
+					To:   otherAddress,
+					Raw: coretypes.Log{
+						BlockNumber: 125,
+					},
+				},
+			},
+			{
+				EventKey: eventlog.ERC721Transfer,
+				Unpacked: struct{}{},
+			},
+		},
+	}
+	pubsub.Publish(transferDetectorPublisher, msg)
+
+	require.Eventually(t, func() bool {
+		return estimatedBlockCalls.Load() >= 3
+	}, 2*time.Second, 50*time.Millisecond)
+}

services/wallet/collectibles/service.go

@@ -212,24 +212,31 @@ func (s *Service) GetOwnedCollectibles(
 	}, err
 }
 
-func (s *Service) needsToFetch(chainID walletCommon.ChainID, address common.Address, fetchCriteria FetchCriteria) (bool, error) {
-	mustFetch := false
+func (s *Service) shouldTriggerLoad(chainID walletCommon.ChainID, address common.Address, fetchCriteria FetchCriteria) (bool, error) {
 	switch fetchCriteria.FetchType {
-	case FetchTypeAlwaysFetch:
-		mustFetch = true
 	case FetchTypeNeverFetch:
-		mustFetch = false
+		return false, nil
+	case FetchTypeAlwaysFetch:
+		return true, nil
 	case FetchTypeFetchIfNotCached, FetchTypeFetchIfCacheOld:
 		timestamp, err := s.ownershipDB.GetOwnershipUpdateTimestamp(address, chainID)
 		if err != nil {
 			return false, err
 		}
-		if timestamp == ownership.InvalidTimestamp ||
-			(fetchCriteria.FetchType == FetchTypeFetchIfCacheOld && timestamp+fetchCriteria.MaxCacheAgeSeconds < time.Now().Unix()) {
-			mustFetch = true
+		if timestamp != ownership.InvalidTimestamp &&
+			!(fetchCriteria.FetchType == FetchTypeFetchIfCacheOld && timestamp+fetchCriteria.MaxCacheAgeSeconds < time.Now().Unix()) {
+			return false, nil
+		}
+		// Skip duplicate load unless the loader is missing or errored
+		if fetchCriteria.FetchType == FetchTypeFetchIfNotCached {
+			state := s.ownershipController.GetLoaderState(chainID, address)
+			if state != ownership.LoaderStateNotAvailable && state != ownership.LoaderStateError {
+				return false, nil
+			}
 		}
+		return true, nil
 	}
-	return mustFetch, nil
+	return false, nil
 }
 
 func (s *Service) fetchOwnedCollectiblesIfNeeded(ctx context.Context, chainIDs []walletCommon.ChainID, addresses []common.Address, fetchCriteria FetchCriteria) error {
@@ -241,21 +248,20 @@ func (s *Service) fetchOwnedCollectiblesIfNeeded(ctx context.Context, chainIDs [
 	wgErr := atomic.NewError(nil)
 	for _, address := range addresses {
 		for _, chainID := range chainIDs {
-			mustFetch, err := s.needsToFetch(chainID, address, fetchCriteria)
-			if err != nil {
+			if ok, err := s.shouldTriggerLoad(chainID, address, fetchCriteria); err != nil {
 				return err
+			} else if !ok {
+				continue
 			}
-			if mustFetch {
-				wg.Add(1)
-				go func() {
-					defer gocommon.LogOnPanic()
-					defer wg.Done()
-					err := s.ownershipController.TriggerLoad(ctx, chainID, address)
-					if err != nil {
-						wgErr.Store(err)
-					}
-				}()
-			}
+
+			wg.Add(1)
+			go func() {
+				defer gocommon.LogOnPanic()
+				defer wg.Done()
+				if err := s.ownershipController.TriggerLoad(ctx, chainID, address); err != nil {
+					wgErr.Store(err)
+				}
+			}()
 		}
 	}