feat: find endpoints asynchronously, cache errors

api/api.go

@@ -125,12 +125,17 @@ func InitServer(ctx context.Context, params APIParams) (Server, error) {
 	if err != nil {
 		return Server{}, errors.Wrap(err, "failed to init lassie")
 	}
-	endpointFinder, err := endpointfinder.NewEndpointFinder(replication.MinerInfoFetcher{
+	infoFetcher := replication.MinerInfoFetcher{
 		Client: util.NewLotusClient(params.LotusAPI, params.LotusToken),
-	}, h, 128)
-	if err != nil {
-		return Server{}, errors.Wrap(err, "failed to init endpoint finder")
 	}
+	endpointFinder := endpointfinder.NewEndpointFinder(
+		infoFetcher,
+		h,
+		endpointfinder.WithLruSize(128),
+		endpointfinder.WithLruTimeout(time.Hour*2),
+		endpointfinder.WithErrorLruSize(128),
+		endpointfinder.WithErrorLruTimeout(time.Minute*5),
+	)
 	return Server{
 		db:          db,
 		host:        h,
@@ -300,7 +305,8 @@ func (s Server) setupRoutes(e *echo.Echo) {
 		db *gorm.DB,
 		storageType string,
 		provider string,
-		request storage.CreateRequest) (*model.Storage, error) {
+		request storage.CreateRequest,
+	) (*model.Storage, error) {
 		request.Provider = provider
 		return s.storageHandler.CreateStorageHandler(ctx, db, storageType, request)
 	}))

retriever/endpointfinder/endpointfinder.go

@@ -7,12 +7,16 @@ import (
 
 	"github.com/data-preservation-programs/singularity/replication"
 	"github.com/filecoin-shipyard/boostly"
-	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/hashicorp/golang-lru/v2/expirable"
+	"github.com/ipfs/go-log/v2"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/peerstore"
+	"go.uber.org/multierr"
 )
 
+var logger = log.Logger("singularity/retriever/endpointfinder")
+
 // ErrHTTPNotSupported indicates we were able to look up the provider and contact them,
 // but they reported that they do not serve HTTP retrievals
 var ErrHTTPNotSupported = errors.New("provider does not support http")
@@ -29,23 +33,26 @@ type MinerInfoFetcher interface {
 type EndpointFinder struct {
 	minerInfoFetcher MinerInfoFetcher
 	h                host.Host
-	httpEndpoints    *lru.Cache[string, []peer.AddrInfo]
+	httpEndpoints    *expirable.LRU[string, []peer.AddrInfo]
+	endpointErrors   *expirable.LRU[string, error]
 }
 
 // NewEndpointFinder returns a new instance of an EndpointFinder
-func NewEndpointFinder(minerInfoFetcher MinerInfoFetcher, h host.Host, size int) (*EndpointFinder, error) {
-	httpEndpoints, err := lru.New[string, []peer.AddrInfo](size)
-	if err != nil {
-		return nil, err
-	}
+func NewEndpointFinder(minerInfoFetcher MinerInfoFetcher, h host.Host, opts ...Option) *EndpointFinder {
+	cfg := applyOptions(opts...)
+
+	httpEndpoints := expirable.NewLRU[string, []peer.AddrInfo](cfg.LruSize, func(key string, value []peer.AddrInfo) {}, cfg.LruTimeout)
+	endpointErrors := expirable.NewLRU[string, error](cfg.ErrorLruSize, func(key string, value error) {}, cfg.ErrorLruTimeout)
+
 	return &EndpointFinder{
 		minerInfoFetcher: minerInfoFetcher,
 		h:                h,
 		httpEndpoints:    httpEndpoints,
-	}, nil
+		endpointErrors:   endpointErrors,
+	}
 }
 
-func (ef *EndpointFinder) fetchHTTPEndpoint(ctx context.Context, provider string) ([]peer.AddrInfo, error) {
+func (ef *EndpointFinder) findHTTPEndpointsForProvider(ctx context.Context, provider string) ([]peer.AddrInfo, error) {
 	// lookup the provider on chain
 	minerInfo, err := ef.minerInfoFetcher.GetProviderInfo(ctx, provider)
 	if err != nil {
@@ -74,30 +81,58 @@ func (ef *EndpointFinder) fetchHTTPEndpoint(ctx context.Context, provider string
 	return nil, ErrHTTPNotSupported
 }
 
-// findOrFetchHTTPEndpoint attempts to load from cache before calling fetchHTTPEndpoint
-func (ef *EndpointFinder) findOrFetchHTTPEndpoint(ctx context.Context, provider string) ([]peer.AddrInfo, error) {
-	addrInfos, has := ef.httpEndpoints.Get(provider)
-	if has {
-		return addrInfos, nil
-	}
-	addrInfos, err := ef.fetchHTTPEndpoint(ctx, provider)
-	if err != nil {
-		return nil, err
-	}
-	ef.httpEndpoints.Add(provider, addrInfos)
-	return addrInfos, nil
-}
-
 // FindHTTPEndpoints finds http endpoints for a given set of providers
 func (ef *EndpointFinder) FindHTTPEndpoints(ctx context.Context, sps []string) ([]peer.AddrInfo, error) {
 	addrInfos := make([]peer.AddrInfo, 0, len(sps))
-	for _, sp := range sps {
-		// TODO: should we ignore if some but not all providers are configured correctly?
-		nextAddrInfos, err := ef.findOrFetchHTTPEndpoint(ctx, sp)
-		if err != nil {
-			return nil, err
+	type findResult struct {
+		addrs []peer.AddrInfo
+		err   error
+	}
+	addrChan := make(chan findResult)
+	var toLookup int
+	var errsum error
+
+	for _, provider := range sps {
+		// first check our caches
+		if providerAddrs, has := ef.httpEndpoints.Get(provider); has {
+			addrInfos = append(addrInfos, providerAddrs...)
+		} else if err, has := ef.endpointErrors.Get(provider); has {
+			logger.Errorf("error looking up http endpoint for %s (cached): %s", provider, err)
+			errsum = multierr.Append(errsum, err)
+		} else {
+			// not in caches, perform full lookup of provider asynchronously
+			toLookup++
+			go func(provider string) {
+				providerAddrs, err := ef.findHTTPEndpointsForProvider(ctx, provider)
+				if err != nil {
+					ef.endpointErrors.Add(provider, err)
+					logger.Errorf("error looking up http endpoint for %s: %s", provider, err)
+				} else {
+					ef.httpEndpoints.Add(provider, providerAddrs)
+				}
+				select {
+				case addrChan <- findResult{addrs: providerAddrs, err: err}:
+				case <-ctx.Done():
+				}
+			}(provider)
+		}
+	}
+
+	for i := 0; i < toLookup; i++ {
+		select {
+		case providerAddrs := <-addrChan:
+			if providerAddrs.addrs != nil {
+				addrInfos = append(addrInfos, providerAddrs.addrs...)
+			} else if providerAddrs.err != nil {
+				errsum = multierr.Append(errsum, providerAddrs.err)
+			}
+		case <-ctx.Done():
+			return nil, ctx.Err()
 		}
-		addrInfos = append(addrInfos, nextAddrInfos...)
+	}
+
+	if len(addrInfos) == 0 {
+		return nil, fmt.Errorf("no http endpoints found for providers %v: %w", sps, errsum)
 	}
 	return addrInfos, nil
 }

retriever/endpointfinder/endpointfinder_test.go

@@ -23,6 +23,7 @@ import (
 func TestEndpointFetcher(t *testing.T) {
 	testCases := []struct {
 		testName                 string
+		providers                int
 		minerInfoNotFindable     bool
 		notDialable              bool
 		notListeningOnTransports bool
@@ -33,27 +34,27 @@ func TestEndpointFetcher(t *testing.T) {
 		{
 			testName:             "unable to find miner on chain",
 			minerInfoNotFindable: true,
-			expectedErrString:    fmt.Errorf("looking up provider info: %w", errMinerNotFound).Error(),
+			expectedErrString:    fmt.Errorf("no http endpoints found for providers [%%s]: looking up provider info: %w", errMinerNotFound).Error(),
 		},
 		{
 			testName:          "unable to dial provider",
 			notDialable:       true,
-			expectedErrString: "querying transports: failed to dial: %s cannot connect to %s",
+			expectedErrString: "no http endpoints found for providers [%s]: querying transports: failed to dial: %s cannot connect to %s",
 		},
 		{
 			testName:                 "provider not listening on protocol",
 			notListeningOnTransports: true,
-			expectedErrString:        "querying transports: failed to negotiate protocol: protocols not supported: [/fil/retrieval/transports/1.0.0]",
+			expectedErrString:        "no http endpoints found for providers [%s]: querying transports: failed to negotiate protocol: protocols not supported: [/fil/retrieval/transports/1.0.0]",
 		},
 		{
 			testName:          "provider not serving http",
 			noHTTP:            true,
-			expectedErrString: endpointfinder.ErrHTTPNotSupported.Error(),
+			expectedErrString: fmt.Errorf("no http endpoints found for providers [%%s]: %w", endpointfinder.ErrHTTPNotSupported).Error(),
 		},
 	}
-	for _, testCase := range testCases {
+	for i, testCase := range testCases {
 		t.Run(testCase.testName, func(t *testing.T) {
-			testProvider := "t01000"
+			testProvider := fmt.Sprintf("t01000%d", i)
 			mn := mocknet.New()
 			source, err := mn.GenPeer()
 			require.NoError(t, err)
@@ -75,7 +76,6 @@ func TestEndpointFetcher(t *testing.T) {
 			require.NoError(t, err)
 			response := boostly.TransportsQueryResponse{}
 			if !testCase.noHTTP {
-
 				response.Protocols = append(response.Protocols, struct {
 					Name      string                `json:"name,omitempty"`
 					Addresses []multiaddr.Multiaddr `json:"addresses,omitempty"`
@@ -90,8 +90,7 @@ func TestEndpointFetcher(t *testing.T) {
 				other.SetStreamHandler(boostly.FilRetrievalTransportsProtocol_1_0_0, handler)
 			}
 
-			endpointFinder, err := endpointfinder.NewEndpointFinder(minerInfoFetcher, source, 3)
-			require.NoError(t, err)
+			endpointFinder := endpointfinder.NewEndpointFinder(minerInfoFetcher, source, endpointfinder.WithErrorLruSize(3), endpointfinder.WithErrorLruSize(3))
 
 			addrInfos, err := endpointFinder.FindHTTPEndpoints(context.Background(), []string{testProvider})
 			if testCase.expectedErrString == "" {
@@ -111,15 +110,15 @@ func TestEndpointFetcher(t *testing.T) {
 				})
 				require.Equal(t, minerInfoFetcher.callCount, 1)
 			} else {
-				errMessage := fmt.Sprintf(testCase.expectedErrString, source.ID(), other.ID())
+				errMessage := fmt.Sprintf(testCase.expectedErrString, testProvider, source.ID(), other.ID())
 				errMessage = strings.Split(errMessage, "%!(EXTRA")[0]
 				require.EqualError(t, err, errMessage)
 				require.Nil(t, addrInfos)
-				// second call should not cache
+				// second call should cache error
 				addrInfos, err := endpointFinder.FindHTTPEndpoints(context.Background(), []string{testProvider})
 				require.EqualError(t, err, errMessage)
 				require.Nil(t, addrInfos)
-				require.Equal(t, minerInfoFetcher.callCount, 2)
+				require.Equal(t, minerInfoFetcher.callCount, 1)
 			}
 		})
 	}

retriever/endpointfinder/options.go

@@ -0,0 +1,55 @@
+package endpointfinder
+
+import "time"
+
+const (
+	defaultLruSize         = 128
+	defaultLruTimeout      = 2 * time.Hour
+	defaultErrorLruSize    = 128
+	defaultErrorLruTimeout = 5 * time.Minute
+)
+
+type config struct {
+	LruSize         int
+	LruTimeout      time.Duration
+	ErrorLruSize    int
+	ErrorLruTimeout time.Duration
+}
+
+func applyOptions(opts ...Option) *config {
+	cfg := &config{
+		LruSize:         defaultLruSize,
+		ErrorLruSize:    defaultErrorLruSize,
+		ErrorLruTimeout: defaultErrorLruTimeout,
+	}
+	for _, opt := range opts {
+		opt(cfg)
+	}
+	return cfg
+}
+
+type Option func(*config)
+
+func WithLruSize(size int) Option {
+	return func(cfg *config) {
+		cfg.LruSize = size
+	}
+}
+
+func WithLruTimeout(timeout time.Duration) Option {
+	return func(cfg *config) {
+		cfg.LruTimeout = timeout
+	}
+}
+
+func WithErrorLruSize(size int) Option {
+	return func(cfg *config) {
+		cfg.ErrorLruSize = size
+	}
+}
+
+func WithErrorLruTimeout(timeout time.Duration) Option {
+	return func(cfg *config) {
+		cfg.ErrorLruTimeout = timeout
+	}
+}